mrleo1nid/stoat-mbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Surface Stoat's voice errors and stop leaving orphan connections

Browse Source 
Joining a channel failed silently: revoice's join() runs an async
executor inside `new Promise`, so a rejected join_call never reaches
reject() — the promise hangs forever and the real error escapes as an
unhandled rejection. The client wrapper now latches API failures and
settles the join itself, translating Stoat's error codes (AlreadyConnected,
LiveKitUnavailable, UnknownNode, ...) into messages the chat can show.

A failed join also used to leave the connection object alive, which kept
the bot registered in the channel and made the next attempt fail with
AlreadyConnected; it is now destroyed on any failure. The LiveKit node
name is configurable via VOICE_NODE for instances that renamed it, and
the README documents how to clear a stuck voice state from Redis.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Leonid Pershin
2026-09-08 23:40:21 +03:00
co-authored by Claude Opus 5
parent d4880e757e
commit 823a9f1565
6 changed files with 584 additions and 464 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+4
View File
@@ -13,6 +13,10 @@ COMMAND_PREFIX=!
# Удалять сообщение с командой после её распознавания (нужно право ManageMessages). # Удалять сообщение с командой после её распознавания (нужно право ManageMessages).
DELETE_COMMAND_MESSAGES=true DELETE_COMMAND_MESSAGES=true
# Имя LiveKit-ноды из Revolt.toml, секция [hosts.livekit].
# В стандартном self-hosted это "worldwide".
VOICE_NODE=worldwide
# ------------------------------------------------------------- Веб-панель --- # ------------------------------------------------------------- Веб-панель ---
PORT=3005 PORT=3005
HOST=0.0.0.0 HOST=0.0.0.0
README.md
+19 -1
View File
@@ -149,7 +149,25 @@ cd /opt/stoat-mbot && docker compose up -d --build && docker compose logs -f
резолвится в хост: `docker compose exec mbot node -e "fetch(process.env.STOAT_API_URL).then(r=>console.log(r.status))"`. резолвится в хост: `docker compose exec mbot node -e "fetch(process.env.STOAT_API_URL).then(r=>console.log(r.status))"`.
2. **`joining voice channel`, но нет `voice connection established`.** Значит `join_call` отдал 2. **`joining voice channel`, но нет `voice connection established`.** Значит `join_call` отдал
токен, а WebSocket до LiveKit не поднялся — смотрите, доступен ли `/livekit` через внешний домен. токен, а WebSocket до LiveKit не поднялся — смотрите, доступен ли `/livekit` через внешний домен.
3. **Бот в канале, но звука нет.** Это уже медиа-трафик: LiveKit анонсирует клиентам свой адрес 3. **`Stoat считает, что бот уже в этом голосовом канале` (`AlreadyConnected`).** Зависшее
состояние в Redis после падения бота: `join_call` регистрирует участника, а выйти он не успел.
Ботам API запрещает `force_disconnect`, поэтому чистим вручную — в каталоге инстанса:
```bash
docker compose exec redis valkey-cli --scan --pattern 'vc:*'
```
```bash
docker compose exec redis valkey-cli DEL 'vc:<ID_бота>'
```
```bash
docker compose exec redis valkey-cli SREM 'vc_members:<ID_голосового_канала>' '<ID_бота>'
```
В норме состояние снимает `voice-ingress` по вебхуку от LiveKit — если ситуация повторяется
после каждого перезапуска, смотрите `docker compose logs voice-ingress`.
4. **Бот в канале, но звука нет.** Это уже медиа-трафик: LiveKit анонсирует клиентам свой адрес
из `rtc.node_ip` / `use_external_ip` в `/opt/stoat/livekit.yml` и ждёт UDP на 50000-50100. из `rtc.node_ip` / `use_external_ip` в `/opt/stoat/livekit.yml` и ждёт UDP на 50000-50100.
Если анонсируется внешний IP, а роутер не умеет NAT loopback, пакеты от контейнера до него не Если анонсируется внешний IP, а роутер не умеет NAT loopback, пакеты от контейнера до него не
дойдут. Тогда либо включите hairpin на роутере, либо запустите бота внутри compose-проекта дойдут. Тогда либо включите hairpin на роутере, либо запустите бота внутри compose-проекта
src/config.ts
+81 -79
View File
@@ -1,79 +1,81 @@
import { readFileSync, existsSync } from "node:fs"; import { readFileSync, existsSync } from "node:fs";
import { z } from "zod"; import { z } from "zod";
// Minimal .env loader so we don't need an extra dependency. // Minimal .env loader so we don't need an extra dependency.
function loadDotEnv(path = ".env"): void { function loadDotEnv(path = ".env"): void {
if (!existsSync(path)) return; if (!existsSync(path)) return;
for (const rawLine of readFileSync(path, "utf8").split(/\r?\n/)) { for (const rawLine of readFileSync(path, "utf8").split(/\r?\n/)) {
const line = rawLine.trim(); const line = rawLine.trim();
if (!line || line.startsWith("#")) continue; if (!line || line.startsWith("#")) continue;
const eq = line.indexOf("="); const eq = line.indexOf("=");
if (eq === -1) continue; if (eq === -1) continue;
const key = line.slice(0, eq).trim(); const key = line.slice(0, eq).trim();
let value = line.slice(eq + 1).trim(); let value = line.slice(eq + 1).trim();
if ( if (
(value.startsWith('"') && value.endsWith('"')) || (value.startsWith('"') && value.endsWith('"')) ||
(value.startsWith("'") && value.endsWith("'")) (value.startsWith("'") && value.endsWith("'"))
) { ) {
value = value.slice(1, -1); value = value.slice(1, -1);
} }
if (process.env[key] === undefined) process.env[key] = value; if (process.env[key] === undefined) process.env[key] = value;
} }
} }
loadDotEnv(); loadDotEnv();
const schema = z.object({ const schema = z.object({
STOAT_API_URL: z.string().url(), STOAT_API_URL: z.string().url(),
STOAT_BOT_TOKEN: z.string().min(1), STOAT_BOT_TOKEN: z.string().min(1),
COMMAND_PREFIX: z.string().min(1).default("!"), COMMAND_PREFIX: z.string().min(1).default("!"),
/** Remove the invoking message after a command is recognised. Needs ManageMessages. */ /** LiveKit node name from Revolt.toml ([hosts.livekit]); self-hosted default is "worldwide". */
DELETE_COMMAND_MESSAGES: z VOICE_NODE: z.string().min(1).default("worldwide"),
.enum(["true", "false"]) /** Remove the invoking message after a command is recognised. Needs ManageMessages. */
.default("true") DELETE_COMMAND_MESSAGES: z
.transform((value) => value === "true"), .enum(["true", "false"])
.default("true")
PORT: z.coerce.number().int().positive().default(3005), .transform((value) => value === "true"),
HOST: z.string().default("0.0.0.0"),
PUBLIC_URL: z.string().url(), PORT: z.coerce.number().int().positive().default(3005),
JWT_SECRET: z.string().min(16), HOST: z.string().default("0.0.0.0"),
SESSION_TTL_HOURS: z.coerce.number().positive().default(168), PUBLIC_URL: z.string().url(),
JWT_SECRET: z.string().min(16),
YTDLP_PATH: z.string().default("yt-dlp"), SESSION_TTL_HOURS: z.coerce.number().positive().default(168),
YTDLP_COOKIES: z.string().optional(),
/** Extra `--extractor-args` values, separated by ";" — e.g. youtube:player_client=default,web_safari */ YTDLP_PATH: z.string().default("yt-dlp"),
YTDLP_EXTRACTOR_ARGS: z.string().optional(), YTDLP_COOKIES: z.string().optional(),
LOCAL_MEDIA_DIR: z.string().optional(), /** Extra `--extractor-args` values, separated by ";" — e.g. youtube:player_client=default,web_safari */
YTDLP_EXTRACTOR_ARGS: z.string().optional(),
DEFAULT_VOLUME: z.coerce.number().min(0).max(200).default(60), LOCAL_MEDIA_DIR: z.string().optional(),
MAX_QUEUE_SIZE: z.coerce.number().int().positive().default(500),
SEARCH_RESULT_LIMIT: z.coerce.number().int().positive().max(25).default(10), DEFAULT_VOLUME: z.coerce.number().min(0).max(200).default(60),
IDLE_TIMEOUT_SECONDS: z.coerce.number().int().min(0).default(300), MAX_QUEUE_SIZE: z.coerce.number().int().positive().default(500),
DJ_ROLE_NAME: z.string().default("DJ"), SEARCH_RESULT_LIMIT: z.coerce.number().int().positive().max(25).default(10),
REQUIRE_DJ_ROLE: z IDLE_TIMEOUT_SECONDS: z.coerce.number().int().min(0).default(300),
.enum(["true", "false"]) DJ_ROLE_NAME: z.string().default("DJ"),
.default("false") REQUIRE_DJ_ROLE: z
.transform((value) => value === "true"), .enum(["true", "false"])
.default("false")
LOG_LEVEL: z.string().default("info"), .transform((value) => value === "true"),
NODE_ENV: z.string().default("development"),
}); LOG_LEVEL: z.string().default("info"),
NODE_ENV: z.string().default("development"),
const parsed = schema.safeParse(process.env); });
if (!parsed.success) { const parsed = schema.safeParse(process.env);
const issues = parsed.error.issues
.map((i) => ` - ${i.path.join(".")}: ${i.message}`) if (!parsed.success) {
.join("\n"); const issues = parsed.error.issues
console.error(`Invalid configuration, check your .env file:\n${issues}`); .map((i) => ` - ${i.path.join(".")}: ${i.message}`)
process.exit(1); .join("\n");
} console.error(`Invalid configuration, check your .env file:\n${issues}`);
process.exit(1);
export const config = { }
...parsed.data,
STOAT_API_URL: parsed.data.STOAT_API_URL.replace(/\/+$/, ""), export const config = {
PUBLIC_URL: parsed.data.PUBLIC_URL.replace(/\/+$/, ""), ...parsed.data,
isProduction: parsed.data.NODE_ENV === "production", STOAT_API_URL: parsed.data.STOAT_API_URL.replace(/\/+$/, ""),
}; PUBLIC_URL: parsed.data.PUBLIC_URL.replace(/\/+$/, ""),
isProduction: parsed.data.NODE_ENV === "production",
export type Config = typeof config; };
export type Config = typeof config;
src/core/manager.ts
+303 -303
View File
@@ -1,303 +1,303 @@
import { EventEmitter } from "node:events"; import { EventEmitter } from "node:events";
import { config } from "../config.js"; import { config } from "../config.js";
import { logger } from "../logger.js"; import { logger } from "../logger.js";
import { resolveQuery, searchTracks } from "../sources/index.js"; import { resolveQuery, searchTracks } from "../sources/index.js";
import { import {
UserFacingError, UserFacingError,
type LoopMode, type LoopMode,
type PlayerSnapshot, type PlayerSnapshot,
type Requester, type Requester,
type SearchResult, type SearchResult,
type Track, type Track,
} from "../types.js"; } from "../types.js";
import { GuildPlayer, type Notice, type PositionUpdate } from "./player.js"; import { GuildPlayer, type Notice, type PositionUpdate } from "./player.js";
import { Revoice, type RevoiceLike } from "./revoice.js"; import { createRevoice, type RevoiceLike } from "./revoice.js";
const log = logger.child({ mod: "manager" }); const log = logger.child({ mod: "manager" });
export interface VoiceChannelRef { export interface VoiceChannelRef {
id: string; id: string;
name: string; name: string;
} }
export interface ServerRef { export interface ServerRef {
id: string; id: string;
name: string; name: string;
iconUrl: string | null; iconUrl: string | null;
} }
/** /**
* Everything the core needs to know about the chat side of Stoat. Implemented on * Everything the core needs to know about the chat side of Stoat. Implemented on
* top of the bot's stoat.js client so the player itself stays testable. * top of the bot's stoat.js client so the player itself stays testable.
*/ */
export interface StoatContext { export interface StoatContext {
findUserVoiceChannel(serverId: string, userId: string): VoiceChannelRef | null; findUserVoiceChannel(serverId: string, userId: string): VoiceChannelRef | null;
getVoiceChannel(channelId: string): VoiceChannelRef | null; getVoiceChannel(channelId: string): VoiceChannelRef | null;
listVoiceChannels(serverId: string): VoiceChannelRef[]; listVoiceChannels(serverId: string): VoiceChannelRef[];
getServerName(serverId: string): string | null; getServerName(serverId: string): string | null;
listServersForUser(userId: string): Promise<ServerRef[]>; listServersForUser(userId: string): Promise<ServerRef[]>;
isMember(serverId: string, userId: string): Promise<boolean>; isMember(serverId: string, userId: string): Promise<boolean>;
canControl(serverId: string, userId: string): Promise<boolean>; canControl(serverId: string, userId: string): Promise<boolean>;
sendMessage(channelId: string, content: string): Promise<void>; sendMessage(channelId: string, content: string): Promise<void>;
} }
export type PlayMode = "append" | "next" | "now"; export type PlayMode = "append" | "next" | "now";
export interface ManagerEvents { export interface ManagerEvents {
update: [PlayerSnapshot]; update: [PlayerSnapshot];
position: [PositionUpdate]; position: [PositionUpdate];
} }
export interface PlayOutcome extends SearchResult { export interface PlayOutcome extends SearchResult {
startedNow: boolean; startedNow: boolean;
queuePosition: number; queuePosition: number;
} }
/** /**
* Owns one GuildPlayer per server and exposes the high-level operations that * Owns one GuildPlayer per server and exposes the high-level operations that
* both the chat commands and the web panel call into. * both the chat commands and the web panel call into.
*/ */
export class MusicManager extends EventEmitter<ManagerEvents> { export class MusicManager extends EventEmitter<ManagerEvents> {
private readonly players = new Map<string, GuildPlayer>(); private readonly players = new Map<string, GuildPlayer>();
private readonly revoice: RevoiceLike; private readonly revoice: RevoiceLike;
private stoat: StoatContext | null = null; private stoat: StoatContext | null = null;
constructor() { constructor() {
super(); super();
this.revoice = new Revoice(config.STOAT_BOT_TOKEN, { baseURL: config.STOAT_API_URL }); this.revoice = createRevoice(config.STOAT_BOT_TOKEN, config.STOAT_API_URL, config.VOICE_NODE);
} }
attachStoat(context: StoatContext): void { attachStoat(context: StoatContext): void {
this.stoat = context; this.stoat = context;
} }
private get chat(): StoatContext { private get chat(): StoatContext {
if (!this.stoat) throw new UserFacingError("Бот ещё не подключился к Stoat"); if (!this.stoat) throw new UserFacingError("Бот ещё не подключился к Stoat");
return this.stoat; return this.stoat;
} }
// ---------------------------------------------------------------- players --- // ---------------------------------------------------------------- players ---
get(serverId: string): GuildPlayer | undefined { get(serverId: string): GuildPlayer | undefined {
return this.players.get(serverId); return this.players.get(serverId);
} }
list(): GuildPlayer[] { list(): GuildPlayer[] {
return [...this.players.values()]; return [...this.players.values()];
} }
getOrCreate(serverId: string): GuildPlayer { getOrCreate(serverId: string): GuildPlayer {
const existing = this.players.get(serverId); const existing = this.players.get(serverId);
if (existing) return existing; if (existing) return existing;
const player = new GuildPlayer({ const player = new GuildPlayer({
serverId, serverId,
serverName: this.stoat?.getServerName(serverId) ?? null, serverName: this.stoat?.getServerName(serverId) ?? null,
revoice: this.revoice, revoice: this.revoice,
}); });
player.on("update", (snapshot) => this.emit("update", snapshot)); player.on("update", (snapshot) => this.emit("update", snapshot));
player.on("position", (position) => this.emit("position", position)); player.on("position", (position) => this.emit("position", position));
player.on("notice", (notice) => void this.deliverNotice(notice)); player.on("notice", (notice) => void this.deliverNotice(notice));
this.players.set(serverId, player); this.players.set(serverId, player);
return player; return player;
} }
private async deliverNotice(notice: Notice): Promise<void> { private async deliverNotice(notice: Notice): Promise<void> {
if (!notice.textChannelId || !this.stoat) return; if (!notice.textChannelId || !this.stoat) return;
try { try {
await this.stoat.sendMessage(notice.textChannelId, notice.text); await this.stoat.sendMessage(notice.textChannelId, notice.text);
} catch (err) { } catch (err) {
log.warn({ err, channel: notice.textChannelId }, "failed to deliver notice"); log.warn({ err, channel: notice.textChannelId }, "failed to deliver notice");
} }
} }
async destroy(serverId: string): Promise<void> { async destroy(serverId: string): Promise<void> {
const player = this.players.get(serverId); const player = this.players.get(serverId);
if (!player) return; if (!player) return;
this.players.delete(serverId); this.players.delete(serverId);
await player.destroy(); await player.destroy();
} }
async destroyAll(): Promise<void> { async destroyAll(): Promise<void> {
await Promise.allSettled([...this.players.keys()].map((id) => this.destroy(id))); await Promise.allSettled([...this.players.keys()].map((id) => this.destroy(id)));
} }
// ------------------------------------------------------------ permissions --- // ------------------------------------------------------------ permissions ---
async assertControl(serverId: string, userId: string): Promise<void> { async assertControl(serverId: string, userId: string): Promise<void> {
if (!(await this.chat.canControl(serverId, userId))) { if (!(await this.chat.canControl(serverId, userId))) {
throw new UserFacingError("Недостаточно прав для управления плеером"); throw new UserFacingError("Недостаточно прав для управления плеером");
} }
} }
// ---------------------------------------------------------------- actions --- // ---------------------------------------------------------------- actions ---
/** Connects to the caller's voice channel (or an explicit one) and returns the player. */ /** Connects to the caller's voice channel (or an explicit one) and returns the player. */
async connect( async connect(
serverId: string, serverId: string,
userId: string, userId: string,
options: { voiceChannelId?: string | null; textChannelId?: string | null } = {}, options: { voiceChannelId?: string | null; textChannelId?: string | null } = {},
): Promise<GuildPlayer> { ): Promise<GuildPlayer> {
const player = this.getOrCreate(serverId); const player = this.getOrCreate(serverId);
if (options.textChannelId) player.textChannelId = options.textChannelId; if (options.textChannelId) player.textChannelId = options.textChannelId;
const target = options.voiceChannelId const target = options.voiceChannelId
? this.chat.getVoiceChannel(options.voiceChannelId) ? this.chat.getVoiceChannel(options.voiceChannelId)
: (this.chat.findUserVoiceChannel(serverId, userId) ?? : (this.chat.findUserVoiceChannel(serverId, userId) ??
(player.voiceChannelId ? this.chat.getVoiceChannel(player.voiceChannelId) : null)); (player.voiceChannelId ? this.chat.getVoiceChannel(player.voiceChannelId) : null));
if (!target) { if (!target) {
throw new UserFacingError("Зайдите в голосовой канал или укажите его явно"); throw new UserFacingError("Зайдите в голосовой канал или укажите его явно");
} }
await player.connect(target.id, target.name); await player.connect(target.id, target.name);
return player; return player;
} }
async play( async play(
serverId: string, serverId: string,
requester: Requester, requester: Requester,
query: string, query: string,
options: { mode?: PlayMode; voiceChannelId?: string | null; textChannelId?: string | null } = {}, options: { mode?: PlayMode; voiceChannelId?: string | null; textChannelId?: string | null } = {},
): Promise<PlayOutcome> { ): Promise<PlayOutcome> {
await this.assertControl(serverId, requester.id); await this.assertControl(serverId, requester.id);
const player = await this.connect(serverId, requester.id, { const player = await this.connect(serverId, requester.id, {
voiceChannelId: options.voiceChannelId ?? null, voiceChannelId: options.voiceChannelId ?? null,
textChannelId: options.textChannelId ?? null, textChannelId: options.textChannelId ?? null,
}); });
const result = await resolveQuery(query, requester, config.MAX_QUEUE_SIZE - player.queue.length); const result = await resolveQuery(query, requester, config.MAX_QUEUE_SIZE - player.queue.length);
if (result.tracks.length === 0) throw new UserFacingError("Ничего не найдено"); if (result.tracks.length === 0) throw new UserFacingError("Ничего не найдено");
const mode = options.mode ?? "append"; const mode = options.mode ?? "append";
const wasIdle = !player.current; const wasIdle = !player.current;
if (mode === "now") { if (mode === "now") {
await player.playNow(result.tracks); await player.playNow(result.tracks);
return { ...result, startedNow: true, queuePosition: 0 }; return { ...result, startedNow: true, queuePosition: 0 };
} }
player.enqueue(result.tracks, mode === "next" ? 0 : undefined); player.enqueue(result.tracks, mode === "next" ? 0 : undefined);
const queuePosition = mode === "next" ? 1 : player.queue.length - result.tracks.length + 1; const queuePosition = mode === "next" ? 1 : player.queue.length - result.tracks.length + 1;
await player.ensurePlaying(); await player.ensurePlaying();
return { ...result, startedNow: wasIdle, queuePosition }; return { ...result, startedNow: wasIdle, queuePosition };
} }
/** Queues already-resolved tracks (used by the panel's search results). */ /** Queues already-resolved tracks (used by the panel's search results). */
async enqueueTracks( async enqueueTracks(
serverId: string, serverId: string,
requester: Requester, requester: Requester,
tracks: Track[], tracks: Track[],
options: { mode?: PlayMode; voiceChannelId?: string | null; textChannelId?: string | null } = {}, options: { mode?: PlayMode; voiceChannelId?: string | null; textChannelId?: string | null } = {},
): Promise<PlayOutcome> { ): Promise<PlayOutcome> {
await this.assertControl(serverId, requester.id); await this.assertControl(serverId, requester.id);
const player = await this.connect(serverId, requester.id, { const player = await this.connect(serverId, requester.id, {
voiceChannelId: options.voiceChannelId ?? null, voiceChannelId: options.voiceChannelId ?? null,
textChannelId: options.textChannelId ?? null, textChannelId: options.textChannelId ?? null,
}); });
const owned = tracks.map((track) => ({ ...track, requestedBy: requester })); const owned = tracks.map((track) => ({ ...track, requestedBy: requester }));
const wasIdle = !player.current; const wasIdle = !player.current;
if (options.mode === "now") { if (options.mode === "now") {
await player.playNow(owned); await player.playNow(owned);
return { tracks: owned, playlist: null, startedNow: true, queuePosition: 0 }; return { tracks: owned, playlist: null, startedNow: true, queuePosition: 0 };
} }
player.enqueue(owned, options.mode === "next" ? 0 : undefined); player.enqueue(owned, options.mode === "next" ? 0 : undefined);
await player.ensurePlaying(); await player.ensurePlaying();
return { return {
tracks: owned, tracks: owned,
playlist: null, playlist: null,
startedNow: wasIdle, startedNow: wasIdle,
queuePosition: options.mode === "next" ? 1 : player.queue.length - owned.length + 1, queuePosition: options.mode === "next" ? 1 : player.queue.length - owned.length + 1,
}; };
} }
search(query: string, requester: Requester, limit?: number): Promise<Track[]> { search(query: string, requester: Requester, limit?: number): Promise<Track[]> {
return searchTracks(query, requester, limit); return searchTracks(query, requester, limit);
} }
private async require(serverId: string, userId: string): Promise<GuildPlayer> { private async require(serverId: string, userId: string): Promise<GuildPlayer> {
await this.assertControl(serverId, userId); await this.assertControl(serverId, userId);
const player = this.players.get(serverId); const player = this.players.get(serverId);
if (!player) throw new UserFacingError("Плеер не запущен на этом сервере"); if (!player) throw new UserFacingError("Плеер не запущен на этом сервере");
return player; return player;
} }
async pause(serverId: string, userId: string): Promise<void> { async pause(serverId: string, userId: string): Promise<void> {
(await this.require(serverId, userId)).pause(); (await this.require(serverId, userId)).pause();
} }
async resume(serverId: string, userId: string): Promise<void> { async resume(serverId: string, userId: string): Promise<void> {
(await this.require(serverId, userId)).resume(); (await this.require(serverId, userId)).resume();
} }
async togglePause(serverId: string, userId: string): Promise<"paused" | "playing"> { async togglePause(serverId: string, userId: string): Promise<"paused" | "playing"> {
const player = await this.require(serverId, userId); const player = await this.require(serverId, userId);
if (player.snapshot().status === "paused") { if (player.snapshot().status === "paused") {
player.resume(); player.resume();
return "playing"; return "playing";
} }
player.pause(); player.pause();
return "paused"; return "paused";
} }
async skip(serverId: string, userId: string, count = 1): Promise<Track | null> { async skip(serverId: string, userId: string, count = 1): Promise<Track | null> {
return (await this.require(serverId, userId)).skip(count); return (await this.require(serverId, userId)).skip(count);
} }
async stop(serverId: string, userId: string): Promise<void> { async stop(serverId: string, userId: string): Promise<void> {
await (await this.require(serverId, userId)).stop(); await (await this.require(serverId, userId)).stop();
} }
async setVolume(serverId: string, userId: string, volume: number): Promise<void> { async setVolume(serverId: string, userId: string, volume: number): Promise<void> {
(await this.require(serverId, userId)).setVolume(volume); (await this.require(serverId, userId)).setVolume(volume);
} }
async setLoop(serverId: string, userId: string, mode: LoopMode): Promise<void> { async setLoop(serverId: string, userId: string, mode: LoopMode): Promise<void> {
(await this.require(serverId, userId)).setLoop(mode); (await this.require(serverId, userId)).setLoop(mode);
} }
async shuffle(serverId: string, userId: string): Promise<void> { async shuffle(serverId: string, userId: string): Promise<void> {
(await this.require(serverId, userId)).shuffle(); (await this.require(serverId, userId)).shuffle();
} }
async seek(serverId: string, userId: string, seconds: number): Promise<void> { async seek(serverId: string, userId: string, seconds: number): Promise<void> {
await (await this.require(serverId, userId)).seek(seconds); await (await this.require(serverId, userId)).seek(seconds);
} }
async remove(serverId: string, userId: string, trackId: string): Promise<Track> { async remove(serverId: string, userId: string, trackId: string): Promise<Track> {
return (await this.require(serverId, userId)).remove(trackId); return (await this.require(serverId, userId)).remove(trackId);
} }
async move(serverId: string, userId: string, trackId: string, toIndex: number): Promise<void> { async move(serverId: string, userId: string, trackId: string, toIndex: number): Promise<void> {
(await this.require(serverId, userId)).move(trackId, toIndex); (await this.require(serverId, userId)).move(trackId, toIndex);
} }
async clearQueue(serverId: string, userId: string): Promise<void> { async clearQueue(serverId: string, userId: string): Promise<void> {
(await this.require(serverId, userId)).clearQueue(); (await this.require(serverId, userId)).clearQueue();
} }
async leave(serverId: string, userId: string): Promise<void> { async leave(serverId: string, userId: string): Promise<void> {
await (await this.require(serverId, userId)).leaveVoice(); await (await this.require(serverId, userId)).leaveVoice();
} }
snapshot(serverId: string): PlayerSnapshot { snapshot(serverId: string): PlayerSnapshot {
const player = this.players.get(serverId); const player = this.players.get(serverId);
if (player) return player.snapshot(); if (player) return player.snapshot();
return { return {
serverId, serverId,
serverName: this.stoat?.getServerName(serverId) ?? null, serverName: this.stoat?.getServerName(serverId) ?? null,
voiceChannelId: null, voiceChannelId: null,
voiceChannelName: null, voiceChannelName: null,
textChannelId: null, textChannelId: null,
status: "idle", status: "idle",
current: null, current: null,
position: 0, position: 0,
queue: [], queue: [],
history: [], history: [],
volume: config.DEFAULT_VOLUME, volume: config.DEFAULT_VOLUME,
loop: "off", loop: "off",
shuffleUsed: false, shuffleUsed: false,
updatedAt: Date.now(), updatedAt: Date.now(),
}; };
} }
} }
src/core/player.ts
+23 -14
View File
@@ -146,20 +146,29 @@ export class GuildPlayer extends EventEmitter<GuildPlayerEvents> {
this.log.info({ channelId }, "joining voice channel"); this.log.info({ channelId }, "joining voice channel");
const connection = await this.revoice.join(channelId); const connection = await this.revoice.join(channelId);
// The room connects asynchronously inside revoice's constructor, so we wait try {
// for it to report readiness rather than polling a state getter. // The room connects asynchronously inside revoice's constructor, so we wait
await new Promise<void>((resolve, reject) => { // for it to report readiness rather than polling a state getter.
const timer = setTimeout( await new Promise<void>((resolve, reject) => {
() => reject(new UserFacingError("Не удалось подключиться к голосовому каналу")), const timer = setTimeout(
JOIN_TIMEOUT_MS, () => reject(new UserFacingError("Не удалось подключиться к голосовому каналу")),
); JOIN_TIMEOUT_MS,
const done = () => { );
clearTimeout(timer); const done = () => {
resolve(); clearTimeout(timer);
}; resolve();
connection.once("join", done); };
connection.once("roomfetched", done); connection.once("join", done);
}); connection.once("roomfetched", done);
});
} catch (err) {
// Leave no orphan: an abandoned connection keeps the bot registered in the
// channel, and Stoat then refuses the next join with AlreadyConnected.
await connection.destroy().catch(() => {});
connection.removeAllListeners();
this.setStatus("idle");
throw err;
}
this.connection = connection; this.connection = connection;
this.voiceReady = true; this.voiceReady = true;
src/core/revoice.ts
+154 -67
View File
@@ -1,67 +1,154 @@
import { createRequire } from "node:module"; import { createRequire } from "node:module";
import type { Readable } from "node:stream"; import type { Readable } from "node:stream";
import { UserFacingError } from "../types.js";
// revoice.js is CommonJS and its bundled typings lag behind the LiveKit rewrite,
// so we load it through require() and describe only the surface we rely on. // revoice.js is CommonJS and its bundled typings lag behind the LiveKit rewrite,
const require = createRequire(import.meta.url); // so we load it through require() and describe only the surface we rely on.
const require = createRequire(import.meta.url);
export interface MediaPlayerLike {
readonly seconds: number; export interface MediaPlayerLike {
readonly duration: number; readonly seconds: number;
codecData?: { duration?: string } | null; readonly duration: number;
paused: boolean; codecData?: { duration?: string } | null;
playing: boolean; paused: boolean;
fProc?: { kill(signal?: string): void } | null; playing: boolean;
originStream?: { destroy(): void } | null; fProc?: { kill(signal?: string): void } | null;
playStream(input: Readable | string, inputOptions?: string[]): Promise<void>; originStream?: { destroy(): void } | null;
pause(): void; playStream(input: Readable | string, inputOptions?: string[]): Promise<void>;
resume(): void; pause(): void;
stop(init?: boolean): void; resume(): void;
destroy(): void; stop(init?: boolean): void;
setVolume(volume: number): void; destroy(): void;
on(event: "start" | "startplay" | "buffer" | "pause" | "unpause" | "finish", listener: () => void): this; setVolume(volume: number): void;
removeAllListeners(event?: string): this; on(event: "start" | "startplay" | "buffer" | "pause" | "unpause" | "finish", listener: () => void): this;
} removeAllListeners(event?: string): this;
}
/** Voice connection state strings emitted by revoice.js (`Revoice.State`). */
export const VOICE_STATE_OFFLINE = "off"; /** Voice connection state strings emitted by revoice.js (`Revoice.State`). */
export const VOICE_STATE_OFFLINE = "off";
export interface VoiceConnectionLike {
channelId: string; export interface VoiceConnectionLike {
play(media: MediaPlayerLike): Promise<void>; channelId: string;
leave(): Promise<void>; play(media: MediaPlayerLike): Promise<void>;
destroy(): Promise<void>; leave(): Promise<void>;
getUsers(): Array<{ id: string }>; destroy(): Promise<void>;
on(event: "join" | "leave" | "roomfetched" | "autoleave", listener: () => void): this; getUsers(): Array<{ id: string }>;
on(event: "state", listener: (state: string) => void): this; on(event: "join" | "leave" | "roomfetched" | "autoleave", listener: () => void): this;
on(event: "userJoin" | "userleave" | "userLeave", listener: (user: { id: string }) => void): this; on(event: "state", listener: (state: string) => void): this;
once(event: "join" | "leave" | "roomfetched", listener: () => void): this; on(event: "userJoin" | "userleave" | "userLeave", listener: (user: { id: string }) => void): this;
removeAllListeners(event?: string): this; once(event: "join" | "leave" | "roomfetched", listener: () => void): this;
} removeAllListeners(event?: string): this;
// NB: revoice.js also exposes `connection.connected` / `isConnected()`, but both }
// call `room.isConnected()` — a getter, not a method, in @livekit/rtc-node 0.13+, // NB: revoice.js also exposes `connection.connected` / `isConnected()`, but both
// so touching them throws a TypeError. GuildPlayer tracks readiness from events. // call `room.isConnected()` — a getter, not a method, in @livekit/rtc-node 0.13+,
// so touching them throws a TypeError. GuildPlayer tracks readiness from events.
export interface RevoiceLike {
join(channelId: string, leaveIfEmpty?: boolean | number): Promise<VoiceConnectionLike>; export interface RevoiceLike {
getVoiceConnection(channelId: string): VoiceConnectionLike | undefined; join(channelId: string, leaveIfEmpty?: boolean | number): Promise<VoiceConnectionLike>;
connections: Map<string, VoiceConnectionLike>; getVoiceConnection(channelId: string): VoiceConnectionLike | undefined;
} connections: Map<string, VoiceConnectionLike>;
}
interface RevoiceModule {
Revoice: new (token: string, apiConfig?: Record<string, unknown>) => RevoiceLike; interface RevoiceModule {
MediaPlayer: new (normalisation?: boolean) => MediaPlayerLike; Revoice: new (token: string, apiConfig?: Record<string, unknown>) => RevoiceLike;
} MediaPlayer: new (normalisation?: boolean) => MediaPlayerLike;
}
const revoice = require("revoice.js") as RevoiceModule;
const revoice = require("revoice.js") as RevoiceModule;
export const Revoice = revoice.Revoice;
export const MediaPlayer = revoice.MediaPlayer; /** How long we wait for Stoat to answer POST /join_call before giving up. */
const JOIN_REQUEST_TIMEOUT_MS = 15_000;
/** Parses ffmpeg's `hh:mm:ss.xx` duration into seconds. */
export function parseFfmpegDuration(value: string | undefined | null): number { export const Revoice = revoice.Revoice;
if (!value) return 0; export const MediaPlayer = revoice.MediaPlayer;
const parts = value.split(":").map((part) => Number.parseFloat(part));
if (parts.some((part) => Number.isNaN(part))) return 0; interface RevoiceInternal extends RevoiceLike {
return parts.reduce((acc, part) => acc * 60 + part, 0); api: { post(path: string, body?: unknown, params?: unknown): Promise<unknown> };
} }
/** Stoat error codes we can explain better than "internal error". */
const API_ERRORS: Record<string, string> = {
AlreadyConnected:
"Stoat считает, что бот уже в этом голосовом канале. Обычно это зависшее состояние после падения — см. README, раздел про AlreadyConnected.",
NotAVoiceChannel: "Это не голосовой канал",
LiveKitUnavailable: "Голосовой сервер (LiveKit) недоступен",
UnknownNode: "LiveKit-нода не найдена в конфигурации инстанса (Revolt.toml, [hosts.livekit])",
CannotJoinCall: "В канале достигнут лимит участников",
MissingPermission: "У бота нет права Connect в этом голосовом канале",
NotFound: "Канал не найден",
};
function describeApiError(err: unknown): Error {
const response = (err as { response?: { status?: number; data?: { type?: string } } }).response;
const type = response?.data?.type;
if (!type) return err as Error;
const message = API_ERRORS[type] ?? `Stoat отклонил запрос: ${type}`;
return new UserFacingError(message);
}
/**
* Wraps the revoice client so that join_call uses the configured LiveKit node
* and API failures surface Stoat's own error code instead of an axios dump.
*/
export function createRevoice(token: string, baseURL: string, node: string): RevoiceLike {
const instance = new Revoice(token, { baseURL }) as RevoiceInternal;
const post = instance.api.post.bind(instance.api);
const join = instance.join.bind(instance);
let pendingError: Error | null = null;
instance.api.post = async (path: string, body?: unknown, params?: unknown) => {
const payload =
path.endsWith("/join_call") && typeof body === "object" && body !== null
? { ...(body as Record<string, unknown>), node }
: body;
try {
return await post(path, payload, params);
} catch (err) {
pendingError = describeApiError(err);
throw pendingError;
}
};
// revoice's join() runs an async executor inside `new Promise`, so a failing
// join_call never reaches its reject() — the promise hangs forever and the
// real error escapes as an unhandled rejection. We latch that error above and
// settle the join ourselves.
instance.join = (channelId: string, leaveIfEmpty?: boolean | number) =>
new Promise<VoiceConnectionLike>((resolve, reject) => {
pendingError = null;
const startedAt = Date.now();
let settled = false;
const settle = (action: () => void) => {
if (settled) return;
settled = true;
clearInterval(watchdog);
action();
};
const watchdog = setInterval(() => {
if (pendingError) {
const error = pendingError;
settle(() => reject(error));
} else if (Date.now() - startedAt > JOIN_REQUEST_TIMEOUT_MS) {
settle(() => reject(new UserFacingError("Stoat не ответил на запрос подключения к каналу")));
}
}, 50);
watchdog.unref?.();
join(channelId, leaveIfEmpty).then(
(connection) => settle(() => resolve(connection)),
(err: unknown) => settle(() => reject(describeApiError(err))),
);
});
return instance;
}
/** Parses ffmpeg's `hh:mm:ss.xx` duration into seconds. */
export function parseFfmpegDuration(value: string | undefined | null): number {
if (!value) return 0;
const parts = value.split(":").map((part) => Number.parseFloat(part));
if (parts.some((part) => Number.isNaN(part))) return 0;
return parts.reduce((acc, part) => acc * 60 + part, 0);
}