mirror of
https://github.com/arthur-pbty/flint.git
synced 2026-08-01 20:29:03 +02:00
feat: améliorer la gestion des bots multiples avec des identifiants uniques et des limites de taux
This commit is contained in:
Puechberty Arthur
+4
-2
@@ -8,7 +8,8 @@ ALLOW_INSECURE_DB_SSL=false
|
||||
PRESENCE_STREAM_URL=https://twitch.tv/discord
|
||||
POSTGRES_DB=discord_bots
|
||||
POSTGRES_USER=discord_bot
|
||||
POSTGRES_PASSWORD=CHANGE_ME_STRONG_PASSWORD
|
||||
POSTGRES_PASSWORD=CHANGE_ME_STRONG_POSTGRES_PASSWORD
|
||||
REDIS_PASSWORD=CHANGE_ME_STRONG_REDIS_PASSWORD
|
||||
POSTGRES_PORT=5432
|
||||
PREFIX=+
|
||||
DEFAULT_LANG=en
|
||||
@@ -18,7 +19,8 @@ LOG_LEVEL=info
|
||||
STATE_BACKEND=memory
|
||||
REDIS_URL=
|
||||
COMMAND_DISPATCH_MODE=local
|
||||
COMMAND_QUEUE_NAME=bot:command-jobs
|
||||
COMMAND_QUEUE_NAME=bot:${botId}:command-jobs
|
||||
GLOBAL_RATE_LIMIT_MAX_REQUESTS=20
|
||||
GLOBAL_RATE_LIMIT_WINDOW_SECONDS=10
|
||||
RATE_LIMIT_FAIL_OPEN=false
|
||||
ENABLE_LEADER_ELECTION=true
|
||||
|
||||
@@ -57,42 +57,32 @@ Professional command framework template for Discord.js `14.26.2` with:
|
||||
- npm run test
|
||||
- npm run check
|
||||
|
||||
## Docker Deployment (Bot + PostgreSQL)
|
||||
## Docker Deployment (2 Bots + PostgreSQL)
|
||||
|
||||
1. Configure `.env` with at least:
|
||||
- `DISCORD_TOKEN`
|
||||
- `DISCORD_CLIENT_ID`
|
||||
1. Fill production env files:
|
||||
- `.env.bot-alpha.prod`
|
||||
- `.env.bot-beta.prod`
|
||||
2. Set strong shared DB credentials:
|
||||
- `POSTGRES_DB`
|
||||
- `POSTGRES_USER`
|
||||
- `POSTGRES_PASSWORD`
|
||||
2. Start stack:
|
||||
3. Start stack:
|
||||
docker compose up -d --build
|
||||
3. Stop stack:
|
||||
4. Stop stack:
|
||||
docker compose down
|
||||
|
||||
By default, `docker-compose.yml` provisions:
|
||||
- `bot`: the Discord bot container
|
||||
- `bot_alpha`: Discord bot instance A
|
||||
- `bot_beta`: Discord bot instance B
|
||||
- `postgres`: PostgreSQL 16 with persistent volume `postgres_data`
|
||||
|
||||
The bot container uses:
|
||||
- `DATABASE_URL=postgresql://<POSTGRES_USER>:<POSTGRES_PASSWORD>@postgres:5432/<POSTGRES_DB>`
|
||||
Both bots use the same PostgreSQL service, and data remains isolated per bot through `bot_id` keys.
|
||||
|
||||
Security defaults:
|
||||
- PostgreSQL is bound to `127.0.0.1` by default in Compose.
|
||||
- Database TLS verification is strict by default when SSL is enabled.
|
||||
- Keep strong values for `POSTGRES_PASSWORD` and rotate credentials if exposed.
|
||||
|
||||
## Multi-Bot With One DB
|
||||
|
||||
You can run several bot services against the same PostgreSQL instance.
|
||||
|
||||
- Keep one shared `postgres` service.
|
||||
- Add additional bot services with different `DISCORD_TOKEN` / `DISCORD_CLIENT_ID`.
|
||||
- Keep `DATABASE_URL` pointing to the same PostgreSQL service.
|
||||
|
||||
Because rows are keyed by `bot_id`, each bot keeps its own independent presence configuration.
|
||||
|
||||
An example with two bot services is available in `docker-compose.multi-bot.example.yml`.
|
||||
- PostgreSQL is exposed only on the internal Compose network (`expose: 5432`, no host bind).
|
||||
- Bot containers enforce `no-new-privileges` and graceful stop.
|
||||
- Log rotation is enabled (`max-size=10m`, `max-file=3`).
|
||||
- Database TLS verification remains strict by default when SSL is enabled.
|
||||
|
||||
## Architecture
|
||||
|
||||
|
||||
@@ -1,49 +0,0 @@
|
||||
services:
|
||||
bot_alpha:
|
||||
build:
|
||||
context: .
|
||||
dockerfile: Dockerfile
|
||||
restart: unless-stopped
|
||||
depends_on:
|
||||
postgres:
|
||||
condition: service_healthy
|
||||
env_file:
|
||||
- .env.bot-alpha
|
||||
environment:
|
||||
DATABASE_URL: postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB}
|
||||
DATABASE_SSL: ${DATABASE_SSL:-false}
|
||||
|
||||
bot_beta:
|
||||
build:
|
||||
context: .
|
||||
dockerfile: Dockerfile
|
||||
restart: unless-stopped
|
||||
depends_on:
|
||||
postgres:
|
||||
condition: service_healthy
|
||||
env_file:
|
||||
- .env.bot-beta
|
||||
environment:
|
||||
DATABASE_URL: postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB}
|
||||
DATABASE_SSL: ${DATABASE_SSL:-false}
|
||||
|
||||
postgres:
|
||||
image: postgres:16-alpine
|
||||
restart: unless-stopped
|
||||
environment:
|
||||
POSTGRES_DB: ${POSTGRES_DB}
|
||||
POSTGRES_USER: ${POSTGRES_USER}
|
||||
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
|
||||
ports:
|
||||
- "127.0.0.1:${POSTGRES_PORT:-5432}:5432"
|
||||
volumes:
|
||||
- postgres_data:/var/lib/postgresql/data
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER} -d ${POSTGRES_DB}"]
|
||||
interval: 10s
|
||||
timeout: 5s
|
||||
retries: 5
|
||||
start_period: 10s
|
||||
|
||||
volumes:
|
||||
postgres_data:
|
||||
+90
-17
@@ -1,35 +1,108 @@
|
||||
x-logging: &default-logging
|
||||
driver: json-file
|
||||
options:
|
||||
max-size: "10m"
|
||||
max-file: "3"
|
||||
|
||||
x-bot-common: &bot-common
|
||||
build:
|
||||
context: .
|
||||
dockerfile: Dockerfile
|
||||
restart: unless-stopped
|
||||
init: true
|
||||
depends_on:
|
||||
postgres:
|
||||
condition: service_healthy
|
||||
redis:
|
||||
condition: service_healthy
|
||||
security_opt:
|
||||
- no-new-privileges:true
|
||||
cap_drop:
|
||||
- ALL
|
||||
stop_grace_period: 30s
|
||||
networks:
|
||||
- edge_net
|
||||
- backend_net
|
||||
logging: *default-logging
|
||||
|
||||
services:
|
||||
bot:
|
||||
build:
|
||||
context: .
|
||||
dockerfile: Dockerfile
|
||||
restart: unless-stopped
|
||||
depends_on:
|
||||
postgres:
|
||||
condition: service_healthy
|
||||
bot_alpha:
|
||||
<<: *bot-common
|
||||
env_file:
|
||||
- .env
|
||||
- .env.bot-alpha.prod
|
||||
environment:
|
||||
DATABASE_URL: postgresql://${POSTGRES_USER}:${POSTGRES_PASSWORD}@postgres:5432/${POSTGRES_DB}
|
||||
DATABASE_SSL: ${DATABASE_SSL:-false}
|
||||
STATE_BACKEND: redis
|
||||
COMMAND_DISPATCH_MODE: local
|
||||
|
||||
bot_beta:
|
||||
<<: *bot-common
|
||||
env_file:
|
||||
- .env.bot-beta.prod
|
||||
environment:
|
||||
STATE_BACKEND: redis
|
||||
COMMAND_DISPATCH_MODE: local
|
||||
|
||||
postgres:
|
||||
image: postgres:16-alpine
|
||||
restart: unless-stopped
|
||||
init: true
|
||||
environment:
|
||||
POSTGRES_DB: ${POSTGRES_DB}
|
||||
POSTGRES_USER: ${POSTGRES_USER}
|
||||
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
|
||||
ports:
|
||||
- "127.0.0.1:${POSTGRES_PORT:-5432}:5432"
|
||||
POSTGRES_DB: discord_bots
|
||||
POSTGRES_USER: discord_bot
|
||||
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?POSTGRES_PASSWORD is required}
|
||||
POSTGRES_INITDB_ARGS: --auth-host=scram-sha-256 --auth-local=scram-sha-256
|
||||
volumes:
|
||||
- postgres_data:/var/lib/postgresql/data
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER} -d ${POSTGRES_DB}"]
|
||||
test: ["CMD-SHELL", "pg_isready -U $$POSTGRES_USER -d $$POSTGRES_DB"]
|
||||
interval: 10s
|
||||
timeout: 5s
|
||||
retries: 5
|
||||
start_period: 15s
|
||||
security_opt:
|
||||
- no-new-privileges:true
|
||||
stop_grace_period: 30s
|
||||
networks:
|
||||
- backend_net
|
||||
logging: *default-logging
|
||||
|
||||
redis:
|
||||
image: redis:7-alpine
|
||||
restart: unless-stopped
|
||||
init: true
|
||||
environment:
|
||||
REDIS_PASSWORD: ${REDIS_PASSWORD:?REDIS_PASSWORD is required}
|
||||
command:
|
||||
- sh
|
||||
- -ec
|
||||
- >
|
||||
exec redis-server
|
||||
--appendonly yes
|
||||
--save 60 1000
|
||||
--requirepass "$${REDIS_PASSWORD}"
|
||||
--maxmemory-policy noeviction
|
||||
volumes:
|
||||
- redis_data:/data
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "redis-cli -a \"$${REDIS_PASSWORD}\" ping | grep -q PONG"]
|
||||
interval: 10s
|
||||
timeout: 5s
|
||||
retries: 5
|
||||
start_period: 10s
|
||||
security_opt:
|
||||
- no-new-privileges:true
|
||||
stop_grace_period: 20s
|
||||
networks:
|
||||
- backend_net
|
||||
logging: *default-logging
|
||||
|
||||
volumes:
|
||||
postgres_data:
|
||||
redis_data:
|
||||
|
||||
networks:
|
||||
edge_net:
|
||||
driver: bridge
|
||||
backend_net:
|
||||
driver: bridge
|
||||
internal: true
|
||||
|
||||
+29
-24
@@ -8,8 +8,8 @@ import { env } from "../config/env.js";
|
||||
import { CommandRegistry } from "../core/commands/registry.js";
|
||||
import { CommandExecutor } from "../core/execution/CommandExecutor.js";
|
||||
import {
|
||||
type CommandDispatchMode,
|
||||
LocalCommandDispatchPort,
|
||||
WorkerCommandDispatchPort,
|
||||
} from "../core/execution/dispatch.js";
|
||||
import {
|
||||
MemoryCooldownStore,
|
||||
@@ -19,7 +19,6 @@ import {
|
||||
MemoryGlobalRateLimitStore,
|
||||
RedisGlobalRateLimitStore,
|
||||
} from "../core/execution/globalRateLimitStore.js";
|
||||
import { RedisCommandJobPublisher } from "../core/execution/redisCommandJobPublisher.js";
|
||||
import { createScopedLogger } from "../core/logging/logger.js";
|
||||
import {
|
||||
LocalLeaderCoordinator,
|
||||
@@ -67,6 +66,11 @@ const bindFatalProcessHandlers = (
|
||||
};
|
||||
|
||||
export const bootstrap = async (): Promise<void> => {
|
||||
const dispatchMode: CommandDispatchMode = env.COMMAND_DISPATCH_MODE;
|
||||
if (dispatchMode === "worker") {
|
||||
throw new Error("Worker mode is not implemented and must not be used in production");
|
||||
}
|
||||
|
||||
const pool = new Pool({
|
||||
connectionString: env.DATABASE_URL,
|
||||
ssl: env.DATABASE_SSL
|
||||
@@ -77,21 +81,32 @@ export const bootstrap = async (): Promise<void> => {
|
||||
: undefined,
|
||||
});
|
||||
|
||||
const requiresRedis = env.STATE_BACKEND === "redis" || env.COMMAND_DISPATCH_MODE === "worker";
|
||||
const requiresRedis = env.STATE_BACKEND === "redis";
|
||||
let redis: Redis | null = null;
|
||||
|
||||
if (requiresRedis) {
|
||||
const redisUrl = env.REDIS_URL;
|
||||
if (!redisUrl) {
|
||||
throw new Error("REDIS_URL is required when STATE_BACKEND=redis or COMMAND_DISPATCH_MODE=worker");
|
||||
throw new Error("REDIS_URL is required when STATE_BACKEND=redis");
|
||||
}
|
||||
|
||||
redis = new Redis(redisUrl, {
|
||||
maxRetriesPerRequest: null,
|
||||
enableReadyCheck: true,
|
||||
});
|
||||
try {
|
||||
redis = new Redis(redisUrl, {
|
||||
maxRetriesPerRequest: null,
|
||||
enableReadyCheck: true,
|
||||
});
|
||||
|
||||
await redis.ping();
|
||||
await redis.ping();
|
||||
} catch (error) {
|
||||
log.error(
|
||||
{
|
||||
err: error,
|
||||
stateBackend: env.STATE_BACKEND,
|
||||
},
|
||||
"redis unavailable at startup",
|
||||
);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
const presenceStore = new PostgresPresenceStore(pool);
|
||||
@@ -127,23 +142,11 @@ export const bootstrap = async (): Promise<void> => {
|
||||
limit: env.GLOBAL_RATE_LIMIT_MAX_REQUESTS,
|
||||
windowSeconds: env.GLOBAL_RATE_LIMIT_WINDOW_SECONDS,
|
||||
},
|
||||
rateLimitFailOpen: env.RATE_LIMIT_FAIL_OPEN,
|
||||
logger: createScopedLogger("command-executor"),
|
||||
});
|
||||
|
||||
const dispatcher = (() => {
|
||||
if (env.COMMAND_DISPATCH_MODE === "worker") {
|
||||
if (!redis) {
|
||||
throw new Error("Worker dispatch mode requires an initialized Redis client");
|
||||
}
|
||||
|
||||
return new WorkerCommandDispatchPort(
|
||||
new RedisCommandJobPublisher(redis, env.COMMAND_QUEUE_NAME),
|
||||
createScopedLogger("command-dispatcher"),
|
||||
);
|
||||
}
|
||||
|
||||
return new LocalCommandDispatchPort(executor);
|
||||
})();
|
||||
const dispatcher = new LocalCommandDispatchPort(executor);
|
||||
|
||||
const leaderCoordinator = env.ENABLE_LEADER_ELECTION
|
||||
? new PostgresLeaderCoordinator(pool, "discordjs-framework-template")
|
||||
@@ -240,7 +243,9 @@ export const bootstrap = async (): Promise<void> => {
|
||||
log.info(
|
||||
{
|
||||
stateBackend: env.STATE_BACKEND,
|
||||
commandDispatchMode: env.COMMAND_DISPATCH_MODE,
|
||||
commandDispatchMode: dispatchMode,
|
||||
configuredCommandDispatchMode: env.COMMAND_DISPATCH_MODE,
|
||||
rateLimitFailOpen: env.RATE_LIMIT_FAIL_OPEN,
|
||||
rateLimit: {
|
||||
maxRequests: env.GLOBAL_RATE_LIMIT_MAX_REQUESTS,
|
||||
windowSeconds: env.GLOBAL_RATE_LIMIT_WINDOW_SECONDS,
|
||||
|
||||
+8
-3
@@ -83,9 +83,14 @@ const envSchema = z.object({
|
||||
STATE_BACKEND: z.enum(["memory", "redis"]).default("memory"),
|
||||
REDIS_URL: z.preprocess(parseOptionalUrl, z.string().url("REDIS_URL must be a valid URL").optional()),
|
||||
COMMAND_DISPATCH_MODE: z.enum(["local", "worker"]).default("local"),
|
||||
COMMAND_QUEUE_NAME: z.string().trim().min(1).default("bot:command-jobs"),
|
||||
COMMAND_QUEUE_NAME: z.string().trim().min(1).default("bot:${botId}:command-jobs"),
|
||||
GLOBAL_RATE_LIMIT_MAX_REQUESTS: z.coerce.number().int().positive().default(20),
|
||||
GLOBAL_RATE_LIMIT_WINDOW_SECONDS: z.coerce.number().int().positive().default(10),
|
||||
RATE_LIMIT_FAIL_OPEN: z
|
||||
.string()
|
||||
.optional()
|
||||
.default("false")
|
||||
.transform(toBoolean),
|
||||
ENABLE_LEADER_ELECTION: z
|
||||
.string()
|
||||
.optional()
|
||||
@@ -101,9 +106,9 @@ if (parsed.DATABASE_SSL && !parsed.DATABASE_SSL_REJECT_UNAUTHORIZED && !parsed.A
|
||||
);
|
||||
}
|
||||
|
||||
if ((parsed.STATE_BACKEND === "redis" || parsed.COMMAND_DISPATCH_MODE === "worker") && !parsed.REDIS_URL) {
|
||||
if (parsed.STATE_BACKEND === "redis" && !parsed.REDIS_URL) {
|
||||
throw new Error(
|
||||
"REDIS_URL is required when STATE_BACKEND=redis or COMMAND_DISPATCH_MODE=worker.",
|
||||
"REDIS_URL is required when STATE_BACKEND=redis.",
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -15,6 +15,7 @@ export interface CommandExecutorDeps {
|
||||
cooldownStore: CooldownStore;
|
||||
globalRateLimitStore: GlobalRateLimitStore;
|
||||
globalRateLimitPolicy: GlobalRateLimitPolicy;
|
||||
rateLimitFailOpen: boolean;
|
||||
logger: AppLogger;
|
||||
}
|
||||
|
||||
@@ -44,7 +45,7 @@ export class CommandExecutor {
|
||||
return;
|
||||
}
|
||||
|
||||
const remainingCooldownSeconds = await this.consumeCooldown(command, ctx.execution.actor.userId);
|
||||
const remainingCooldownSeconds = await this.consumeCooldown(command, ctx);
|
||||
if (remainingCooldownSeconds > 0) {
|
||||
await ctx.reply(ctx.t("errors.cooldown", { seconds: remainingCooldownSeconds }));
|
||||
return;
|
||||
@@ -128,30 +129,60 @@ export class CommandExecutor {
|
||||
.trim();
|
||||
}
|
||||
|
||||
private async consumeCooldown(command: BotCommand, userId: string): Promise<number> {
|
||||
private async consumeCooldown(command: BotCommand, ctx: CommandExecutionContext): Promise<number> {
|
||||
if (command.cooldown === undefined || command.cooldown <= 0) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
const botId = this.resolveBotId(ctx);
|
||||
if (!botId) {
|
||||
return this.deps.rateLimitFailOpen ? 0 : Math.max(1, command.cooldown);
|
||||
}
|
||||
|
||||
const userId = ctx.execution.actor.userId;
|
||||
|
||||
try {
|
||||
const result = await this.deps.cooldownStore.consume(command.meta.name, userId, command.cooldown);
|
||||
const result = await this.deps.cooldownStore.consume(botId, command.meta.name, userId, command.cooldown);
|
||||
return result.allowed ? 0 : result.retryAfterSeconds;
|
||||
} catch (error) {
|
||||
this.deps.logger.warn(
|
||||
if (this.deps.rateLimitFailOpen) {
|
||||
this.deps.logger.error(
|
||||
{
|
||||
requestId: ctx.execution.requestId,
|
||||
command: command.meta.name,
|
||||
userId,
|
||||
botId,
|
||||
err: error,
|
||||
},
|
||||
"cooldown store unavailable, fail-open enabled",
|
||||
);
|
||||
return 0;
|
||||
}
|
||||
|
||||
this.deps.logger.error(
|
||||
{
|
||||
requestId: ctx.execution.requestId,
|
||||
command: command.meta.name,
|
||||
userId,
|
||||
botId,
|
||||
err: error,
|
||||
},
|
||||
"cooldown store unavailable, continuing without cooldown enforcement",
|
||||
"cooldown store unavailable, fail-closed blocking command",
|
||||
);
|
||||
return 0;
|
||||
|
||||
return Math.max(1, command.cooldown);
|
||||
}
|
||||
}
|
||||
|
||||
private async consumeGlobalRateLimit(ctx: CommandExecutionContext): Promise<number> {
|
||||
const botId = this.resolveBotId(ctx);
|
||||
if (!botId) {
|
||||
return this.deps.rateLimitFailOpen ? 0 : Math.max(1, this.deps.globalRateLimitPolicy.windowSeconds);
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await this.deps.globalRateLimitStore.consume(
|
||||
botId,
|
||||
ctx.execution.actor.userId,
|
||||
this.deps.globalRateLimitPolicy,
|
||||
);
|
||||
@@ -160,21 +191,63 @@ export class CommandExecutor {
|
||||
return 0;
|
||||
}
|
||||
|
||||
return result.retryAfterSeconds;
|
||||
} catch (error) {
|
||||
this.deps.logger.warn(
|
||||
{
|
||||
requestId: ctx.execution.requestId,
|
||||
userId: ctx.execution.actor.userId,
|
||||
err: error,
|
||||
botId,
|
||||
retryAfterSeconds: result.retryAfterSeconds,
|
||||
remaining: result.remaining,
|
||||
limit: result.limit,
|
||||
},
|
||||
"global rate limit store unavailable, continuing without rate limiting",
|
||||
"global rate limit blocked command",
|
||||
);
|
||||
|
||||
return 0;
|
||||
return result.retryAfterSeconds;
|
||||
} catch (error) {
|
||||
if (this.deps.rateLimitFailOpen) {
|
||||
this.deps.logger.error(
|
||||
{
|
||||
requestId: ctx.execution.requestId,
|
||||
userId: ctx.execution.actor.userId,
|
||||
botId,
|
||||
err: error,
|
||||
},
|
||||
"global rate limit store unavailable, fail-open enabled",
|
||||
);
|
||||
return 0;
|
||||
}
|
||||
|
||||
this.deps.logger.error(
|
||||
{
|
||||
requestId: ctx.execution.requestId,
|
||||
userId: ctx.execution.actor.userId,
|
||||
botId,
|
||||
err: error,
|
||||
},
|
||||
"global rate limit store unavailable, fail-closed blocking command",
|
||||
);
|
||||
|
||||
return Math.max(1, this.deps.globalRateLimitPolicy.windowSeconds);
|
||||
}
|
||||
}
|
||||
|
||||
private resolveBotId(ctx: CommandExecutionContext): string | null {
|
||||
const botId = ctx.transport.client.user?.id;
|
||||
if (!botId) {
|
||||
this.deps.logger.error(
|
||||
{
|
||||
requestId: ctx.execution.requestId,
|
||||
source: ctx.execution.source,
|
||||
},
|
||||
"runtime bot id unavailable for scoped execution stores",
|
||||
);
|
||||
return null;
|
||||
}
|
||||
|
||||
return botId;
|
||||
}
|
||||
|
||||
private isSensitiveCommand(command: BotCommand): boolean {
|
||||
return command.sensitive || command.permissions.length > 0;
|
||||
}
|
||||
|
||||
@@ -9,19 +9,24 @@ export interface CooldownConsumeResult {
|
||||
}
|
||||
|
||||
export interface CooldownStore {
|
||||
consume(commandName: string, subjectId: string, cooldownSeconds: number): Promise<CooldownConsumeResult>;
|
||||
consume(botId: string, commandName: string, subjectId: string, cooldownSeconds: number): Promise<CooldownConsumeResult>;
|
||||
}
|
||||
|
||||
export class MemoryCooldownStore implements CooldownStore {
|
||||
private readonly cooldowns = new Map<string, number>();
|
||||
private lastSweepAt = 0;
|
||||
|
||||
public async consume(commandName: string, subjectId: string, cooldownSeconds: number): Promise<CooldownConsumeResult> {
|
||||
public async consume(
|
||||
botId: string,
|
||||
commandName: string,
|
||||
subjectId: string,
|
||||
cooldownSeconds: number,
|
||||
): Promise<CooldownConsumeResult> {
|
||||
if (cooldownSeconds <= 0) {
|
||||
return { allowed: true, retryAfterSeconds: 0 };
|
||||
}
|
||||
|
||||
const key = this.key(commandName, subjectId);
|
||||
const key = this.key(botId, commandName, subjectId);
|
||||
const now = Date.now();
|
||||
this.sweepExpired(now);
|
||||
|
||||
@@ -37,8 +42,8 @@ export class MemoryCooldownStore implements CooldownStore {
|
||||
return { allowed: true, retryAfterSeconds: 0 };
|
||||
}
|
||||
|
||||
private key(commandName: string, subjectId: string): string {
|
||||
return `${commandName}:${subjectId}`;
|
||||
private key(botId: string, commandName: string, subjectId: string): string {
|
||||
return `${botId}:${commandName}:${subjectId}`;
|
||||
}
|
||||
|
||||
private sweepExpired(now: number): void {
|
||||
@@ -65,15 +70,20 @@ export class MemoryCooldownStore implements CooldownStore {
|
||||
export class RedisCooldownStore implements CooldownStore {
|
||||
public constructor(
|
||||
private readonly redis: Redis,
|
||||
private readonly keyPrefix = "bot:cooldown",
|
||||
private readonly keyPrefix = "bot",
|
||||
) {}
|
||||
|
||||
public async consume(commandName: string, subjectId: string, cooldownSeconds: number): Promise<CooldownConsumeResult> {
|
||||
public async consume(
|
||||
botId: string,
|
||||
commandName: string,
|
||||
subjectId: string,
|
||||
cooldownSeconds: number,
|
||||
): Promise<CooldownConsumeResult> {
|
||||
if (cooldownSeconds <= 0) {
|
||||
return { allowed: true, retryAfterSeconds: 0 };
|
||||
}
|
||||
|
||||
const key = this.key(commandName, subjectId);
|
||||
const key = this.key(botId, commandName, subjectId);
|
||||
const result = await this.redis.set(key, "1", "EX", cooldownSeconds, "NX");
|
||||
if (result === "OK") {
|
||||
return {
|
||||
@@ -82,14 +92,19 @@ export class RedisCooldownStore implements CooldownStore {
|
||||
};
|
||||
}
|
||||
|
||||
const ttl = await this.redis.ttl(key);
|
||||
let ttl = await this.redis.ttl(key);
|
||||
if (ttl <= 0) {
|
||||
await this.redis.expire(key, cooldownSeconds);
|
||||
ttl = cooldownSeconds;
|
||||
}
|
||||
|
||||
return {
|
||||
allowed: false,
|
||||
retryAfterSeconds: ttl > 0 ? ttl : cooldownSeconds,
|
||||
retryAfterSeconds: ttl,
|
||||
};
|
||||
}
|
||||
|
||||
private key(commandName: string, subjectId: string): string {
|
||||
return `${this.keyPrefix}:${commandName}:${subjectId}`;
|
||||
private key(botId: string, commandName: string, subjectId: string): string {
|
||||
return `${this.keyPrefix}:${botId}:cooldown:${commandName}:${subjectId}`;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -15,6 +15,7 @@ export interface CommandDispatchPort {
|
||||
export type CommandDispatchMode = "local" | "worker";
|
||||
|
||||
export interface CommandExecutionJob {
|
||||
botId: string;
|
||||
requestId: string;
|
||||
commandName: string;
|
||||
source: CommandSource;
|
||||
@@ -65,11 +66,17 @@ export class WorkerCommandDispatchPort implements CommandDispatchPort {
|
||||
}
|
||||
|
||||
const toExecutionJob = (command: BotCommand, ctx: CommandExecutionContext): CommandExecutionJob => {
|
||||
const botId = ctx.transport.client.user?.id;
|
||||
if (!botId) {
|
||||
throw new Error("runtime bot id unavailable for worker command dispatch");
|
||||
}
|
||||
|
||||
const serializedArgs = Object.fromEntries(
|
||||
Object.entries(ctx.execution.args).map(([key, value]) => [key, serializeArgValue(value)]),
|
||||
);
|
||||
|
||||
return {
|
||||
botId,
|
||||
requestId: ctx.execution.requestId,
|
||||
commandName: command.meta.name,
|
||||
source: ctx.execution.source,
|
||||
|
||||
@@ -13,7 +13,7 @@ export interface GlobalRateLimitDecision {
|
||||
}
|
||||
|
||||
export interface GlobalRateLimitStore {
|
||||
consume(userId: string, policy: GlobalRateLimitPolicy): Promise<GlobalRateLimitDecision>;
|
||||
consume(botId: string, userId: string, policy: GlobalRateLimitPolicy): Promise<GlobalRateLimitDecision>;
|
||||
}
|
||||
|
||||
interface WindowCounterEntry {
|
||||
@@ -24,10 +24,10 @@ interface WindowCounterEntry {
|
||||
export class MemoryGlobalRateLimitStore implements GlobalRateLimitStore {
|
||||
private readonly counters = new Map<string, WindowCounterEntry>();
|
||||
|
||||
public async consume(userId: string, policy: GlobalRateLimitPolicy): Promise<GlobalRateLimitDecision> {
|
||||
public async consume(botId: string, userId: string, policy: GlobalRateLimitPolicy): Promise<GlobalRateLimitDecision> {
|
||||
const sanitized = sanitizePolicy(policy);
|
||||
const now = Date.now();
|
||||
const key = this.key(userId, sanitized.windowSeconds);
|
||||
const key = this.key(botId, userId, sanitized.windowSeconds);
|
||||
const existing = this.counters.get(key);
|
||||
|
||||
if (!existing || existing.resetAt <= now) {
|
||||
@@ -58,28 +58,51 @@ export class MemoryGlobalRateLimitStore implements GlobalRateLimitStore {
|
||||
};
|
||||
}
|
||||
|
||||
private key(userId: string, windowSeconds: number): string {
|
||||
return `${userId}:${windowSeconds}`;
|
||||
private key(botId: string, userId: string, windowSeconds: number): string {
|
||||
return `${botId}:${userId}:${windowSeconds}`;
|
||||
}
|
||||
}
|
||||
|
||||
export class RedisGlobalRateLimitStore implements GlobalRateLimitStore {
|
||||
private static readonly CONSUME_WINDOW_SCRIPT = `
|
||||
local key = KEYS[1]
|
||||
local window = tonumber(ARGV[1])
|
||||
|
||||
if not window or window <= 0 then
|
||||
return redis.error_reply("invalid window")
|
||||
end
|
||||
|
||||
local count = redis.call("INCR", key)
|
||||
if count == 1 then
|
||||
redis.call("EXPIRE", key, window)
|
||||
return { count, window }
|
||||
end
|
||||
|
||||
local ttl = redis.call("TTL", key)
|
||||
if ttl < 0 then
|
||||
redis.call("EXPIRE", key, window)
|
||||
ttl = window
|
||||
end
|
||||
|
||||
return { count, ttl }
|
||||
`;
|
||||
|
||||
public constructor(
|
||||
private readonly redis: Redis,
|
||||
private readonly keyPrefix = "bot:ratelimit:user",
|
||||
private readonly keyPrefix = "bot",
|
||||
) {}
|
||||
|
||||
public async consume(userId: string, policy: GlobalRateLimitPolicy): Promise<GlobalRateLimitDecision> {
|
||||
public async consume(botId: string, userId: string, policy: GlobalRateLimitPolicy): Promise<GlobalRateLimitDecision> {
|
||||
const sanitized = sanitizePolicy(policy);
|
||||
const key = this.key(userId);
|
||||
const key = this.key(botId, userId);
|
||||
|
||||
const count = await this.redis.incr(key);
|
||||
if (count === 1) {
|
||||
await this.redis.expire(key, sanitized.windowSeconds);
|
||||
}
|
||||
|
||||
const ttl = await this.redis.ttl(key);
|
||||
const retryAfterSeconds = ttl > 0 ? ttl : sanitized.windowSeconds;
|
||||
const rawResult = await this.redis.eval(
|
||||
RedisGlobalRateLimitStore.CONSUME_WINDOW_SCRIPT,
|
||||
1,
|
||||
key,
|
||||
String(sanitized.windowSeconds),
|
||||
);
|
||||
const { count, retryAfterSeconds } = parseScriptResult(rawResult, sanitized.windowSeconds);
|
||||
const allowed = count <= sanitized.limit;
|
||||
|
||||
return {
|
||||
@@ -90,8 +113,8 @@ export class RedisGlobalRateLimitStore implements GlobalRateLimitStore {
|
||||
};
|
||||
}
|
||||
|
||||
private key(userId: string): string {
|
||||
return `${this.keyPrefix}:${userId}`;
|
||||
private key(botId: string, userId: string): string {
|
||||
return `${this.keyPrefix}:${botId}:ratelimit:user:${userId}`;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -106,3 +129,36 @@ const sanitizePolicy = (policy: GlobalRateLimitPolicy): GlobalRateLimitPolicy =>
|
||||
windowSeconds,
|
||||
};
|
||||
};
|
||||
|
||||
const parseScriptResult = (rawResult: unknown, fallbackWindowSeconds: number): { count: number; retryAfterSeconds: number } => {
|
||||
if (!Array.isArray(rawResult) || rawResult.length < 2) {
|
||||
throw new Error("Redis rate limit script returned an unexpected payload");
|
||||
}
|
||||
|
||||
const count = toPositiveInt(rawResult[0]);
|
||||
if (!count) {
|
||||
throw new Error("Redis rate limit script returned an invalid counter value");
|
||||
}
|
||||
|
||||
const ttl = toPositiveInt(rawResult[1]) ?? fallbackWindowSeconds;
|
||||
|
||||
return {
|
||||
count,
|
||||
retryAfterSeconds: ttl,
|
||||
};
|
||||
};
|
||||
|
||||
const toPositiveInt = (value: unknown): number | null => {
|
||||
const numeric = typeof value === "number"
|
||||
? value
|
||||
: typeof value === "string"
|
||||
? Number(value)
|
||||
: Number.NaN;
|
||||
|
||||
if (!Number.isFinite(numeric)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const integer = Math.floor(numeric);
|
||||
return integer > 0 ? integer : null;
|
||||
};
|
||||
|
||||
@@ -5,10 +5,26 @@ import type { CommandExecutionJob, CommandJobPublisher } from "./dispatch.js";
|
||||
export class RedisCommandJobPublisher implements CommandJobPublisher {
|
||||
public constructor(
|
||||
private readonly redis: Redis,
|
||||
private readonly queueName = "bot:command-jobs",
|
||||
private readonly queueName = "bot:${botId}:command-jobs",
|
||||
) {}
|
||||
|
||||
public async publish(job: CommandExecutionJob): Promise<void> {
|
||||
await this.redis.lpush(this.queueName, JSON.stringify(job));
|
||||
await this.redis.lpush(this.resolveQueueName(job.botId), JSON.stringify(job));
|
||||
}
|
||||
|
||||
private resolveQueueName(botId: string): string {
|
||||
if (this.queueName.includes("${botId}")) {
|
||||
return this.queueName.replaceAll("${botId}", botId);
|
||||
}
|
||||
|
||||
if (this.queueName === "bot:command-jobs") {
|
||||
return `bot:${botId}:command-jobs`;
|
||||
}
|
||||
|
||||
if (this.queueName.startsWith("bot:")) {
|
||||
return this.queueName;
|
||||
}
|
||||
|
||||
return `bot:${botId}:${this.queueName}`;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,11 +1,11 @@
|
||||
import type { Pool } from "pg";
|
||||
|
||||
export interface LeaderCoordinator {
|
||||
runIfLeader(lockName: string, task: () => Promise<void>): Promise<boolean>;
|
||||
runIfLeader(lockName: string, botId: string, task: () => Promise<void>): Promise<boolean>;
|
||||
}
|
||||
|
||||
export class LocalLeaderCoordinator implements LeaderCoordinator {
|
||||
public async runIfLeader(_lockName: string, task: () => Promise<void>): Promise<boolean> {
|
||||
public async runIfLeader(_lockName: string, _botId: string, task: () => Promise<void>): Promise<boolean> {
|
||||
await task();
|
||||
return true;
|
||||
}
|
||||
@@ -17,8 +17,8 @@ export class PostgresLeaderCoordinator implements LeaderCoordinator {
|
||||
private readonly namespace = "discord-bot",
|
||||
) {}
|
||||
|
||||
public async runIfLeader(lockName: string, task: () => Promise<void>): Promise<boolean> {
|
||||
const advisoryLockKey = hashToInt32(`${this.namespace}:${lockName}`);
|
||||
public async runIfLeader(lockName: string, botId: string, task: () => Promise<void>): Promise<boolean> {
|
||||
const advisoryLockKey = hashToInt32(`${this.namespace}:bot:${botId}:leader:${lockName}`);
|
||||
const client = await this.pool.connect();
|
||||
|
||||
try {
|
||||
|
||||
+9
-2
@@ -19,8 +19,15 @@ export const registerClientReady = (
|
||||
): void => {
|
||||
client.once(Events.ClientReady, async () => {
|
||||
log.info({ botTag: client.user?.tag ?? "unknown" }, "client ready");
|
||||
|
||||
const botId = client.user?.id;
|
||||
if (!botId) {
|
||||
log.error("client ready event received without bot id, leader-only tasks skipped");
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const restoredByLeader = await leaderCoordinator.runIfLeader("presence-restore", async () => {
|
||||
const restoredByLeader = await leaderCoordinator.runIfLeader("presence-restore", botId, async () => {
|
||||
await presenceService.restoreFromStorage(client);
|
||||
});
|
||||
|
||||
@@ -33,7 +40,7 @@ export const registerClientReady = (
|
||||
|
||||
if (env.AUTO_DEPLOY_SLASH) {
|
||||
try {
|
||||
const deployedByLeader = await leaderCoordinator.runIfLeader("slash-deploy", async () => {
|
||||
const deployedByLeader = await leaderCoordinator.runIfLeader("slash-deploy", botId, async () => {
|
||||
const result = await deployApplicationCommands({
|
||||
token: env.DISCORD_TOKEN,
|
||||
clientId: env.DISCORD_CLIENT_ID,
|
||||
|
||||
@@ -12,6 +12,36 @@ import {
|
||||
} from "./commandExecutionContext.js";
|
||||
import { createPrefixReply } from "./replyAdapter.js";
|
||||
|
||||
const PREFIX_PRE_PARSE_THROTTLE_MS = 300;
|
||||
const PREFIX_PRE_PARSE_THROTTLE_SWEEP_INTERVAL_MS = 60_000;
|
||||
const PREFIX_PRE_PARSE_THROTTLE_MAX_AGE_MS = 5 * 60_000;
|
||||
|
||||
const prefixPreParseThrottle = new Map<string, number>();
|
||||
let prefixPreParseThrottleLastSweepAt = 0;
|
||||
|
||||
const shouldThrottlePrefixPreParse = (userId: string): boolean => {
|
||||
const now = Date.now();
|
||||
const lastSeenAt = prefixPreParseThrottle.get(userId);
|
||||
|
||||
if (lastSeenAt !== undefined && now - lastSeenAt < PREFIX_PRE_PARSE_THROTTLE_MS) {
|
||||
return true;
|
||||
}
|
||||
|
||||
prefixPreParseThrottle.set(userId, now);
|
||||
|
||||
if (now - prefixPreParseThrottleLastSweepAt >= PREFIX_PRE_PARSE_THROTTLE_SWEEP_INTERVAL_MS) {
|
||||
for (const [trackedUserId, trackedAt] of prefixPreParseThrottle.entries()) {
|
||||
if (now - trackedAt >= PREFIX_PRE_PARSE_THROTTLE_MAX_AGE_MS) {
|
||||
prefixPreParseThrottle.delete(trackedUserId);
|
||||
}
|
||||
}
|
||||
|
||||
prefixPreParseThrottleLastSweepAt = now;
|
||||
}
|
||||
|
||||
return false;
|
||||
};
|
||||
|
||||
const resolvePrefixLang = (
|
||||
deps: PrefixHandlerDeps,
|
||||
command: BotCommand,
|
||||
@@ -50,6 +80,10 @@ export const createPrefixHandler = (deps: PrefixHandlerDeps) => {
|
||||
return;
|
||||
}
|
||||
|
||||
if (shouldThrottlePrefixPreParse(message.author.id)) {
|
||||
return;
|
||||
}
|
||||
|
||||
const reply = createPrefixReply(message);
|
||||
|
||||
const command = match.command;
|
||||
|
||||
@@ -7,15 +7,27 @@ import { MemoryGlobalRateLimitStore } from "../src/core/execution/globalRateLimi
|
||||
test("MemoryCooldownStore bloque les executions pendant la fenetre", async () => {
|
||||
const store = new MemoryCooldownStore();
|
||||
|
||||
const first = await store.consume("ping", "user-1", 10);
|
||||
const first = await store.consume("bot-1", "ping", "user-1", 10);
|
||||
assert.equal(first.allowed, true);
|
||||
assert.equal(first.retryAfterSeconds, 0);
|
||||
|
||||
const second = await store.consume("ping", "user-1", 10);
|
||||
const second = await store.consume("bot-1", "ping", "user-1", 10);
|
||||
assert.equal(second.allowed, false);
|
||||
assert.ok(second.retryAfterSeconds > 0);
|
||||
});
|
||||
|
||||
test("MemoryCooldownStore isole les cooldowns par bot", async () => {
|
||||
const store = new MemoryCooldownStore();
|
||||
|
||||
const firstBot = await store.consume("bot-a", "ping", "user-1", 10);
|
||||
const secondBot = await store.consume("bot-b", "ping", "user-1", 10);
|
||||
const sameBotAgain = await store.consume("bot-a", "ping", "user-1", 10);
|
||||
|
||||
assert.equal(firstBot.allowed, true);
|
||||
assert.equal(secondBot.allowed, true);
|
||||
assert.equal(sameBotAgain.allowed, false);
|
||||
});
|
||||
|
||||
test("MemoryGlobalRateLimitStore applique la limite globale utilisateur", async () => {
|
||||
const store = new MemoryGlobalRateLimitStore();
|
||||
const policy = {
|
||||
@@ -23,15 +35,31 @@ test("MemoryGlobalRateLimitStore applique la limite globale utilisateur", async
|
||||
windowSeconds: 30,
|
||||
};
|
||||
|
||||
const first = await store.consume("user-rl", policy);
|
||||
const first = await store.consume("bot-1", "user-rl", policy);
|
||||
assert.equal(first.allowed, true);
|
||||
assert.equal(first.remaining, 1);
|
||||
|
||||
const second = await store.consume("user-rl", policy);
|
||||
const second = await store.consume("bot-1", "user-rl", policy);
|
||||
assert.equal(second.allowed, true);
|
||||
assert.equal(second.remaining, 0);
|
||||
|
||||
const third = await store.consume("user-rl", policy);
|
||||
const third = await store.consume("bot-1", "user-rl", policy);
|
||||
assert.equal(third.allowed, false);
|
||||
assert.ok(third.retryAfterSeconds > 0);
|
||||
});
|
||||
|
||||
test("MemoryGlobalRateLimitStore isole les compteurs par bot", async () => {
|
||||
const store = new MemoryGlobalRateLimitStore();
|
||||
const policy = {
|
||||
limit: 1,
|
||||
windowSeconds: 30,
|
||||
};
|
||||
|
||||
const botAFirst = await store.consume("bot-a", "user-rl", policy);
|
||||
const botBFirst = await store.consume("bot-b", "user-rl", policy);
|
||||
const botASecond = await store.consume("bot-a", "user-rl", policy);
|
||||
|
||||
assert.equal(botAFirst.allowed, true);
|
||||
assert.equal(botBFirst.allowed, true);
|
||||
assert.equal(botASecond.allowed, false);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user