From ff078fc2543ed38e6a40c58b4216b5979745bebe Mon Sep 17 00:00:00 2001 From: Bianca Date: Mon, 26 Oct 2026 20:58:59 +0100 Subject: [PATCH] feat(groups): answer the backend with each group's native admins People needs who administers each Telegram group, and Telegram is the only source (RFC v3 decision 2). The bot keeps a cache of getChatAdministrators (without bots) in Redis and answers the backend's getGroupAdmins over the socket; the backend stores nothing. The cache is refreshed when someone becomes or stops being an admin, once a burst of changes settles; 30 seconds after the bot is promoted in a group; and for every group 2 minutes after startup and every 6 hours, one Telegram request at a time. Entries expire after a day, so a group the bot can no longer read becomes unknown instead of stale. Needs @polinetwork/backend with getGroupAdmins (PoliNetworkOrg/backend#98). --- src/bot.ts | 2 + src/lib/group-admins/index.ts | 102 ++++++++++++++++++++++++ src/middlewares/group-admins-sync.ts | 55 +++++++++++++ src/modules/websocket.ts | 5 ++ tests/group-admins.test.ts | 115 +++++++++++++++++++++++++++ 5 files changed, 279 insertions(+) create mode 100644 src/lib/group-admins/index.ts create mode 100644 src/middlewares/group-admins-sync.ts create mode 100644 tests/group-admins.test.ts diff --git a/src/bot.ts b/src/bot.ts index 71ccc36..7d75273 100644 --- a/src/bot.ts +++ b/src/bot.ts @@ -14,6 +14,7 @@ import { AutoModerationStack } from "./middlewares/auto-moderation-stack" import { BotMembershipHandler } from "./middlewares/bot-membership-handler" import { CampaignSpamGuard } from "./middlewares/campaign-spam" import { checkUsername } from "./middlewares/check-username" +import { GroupAdminsSync } from "./middlewares/group-admins-sync" import { GroupSpecificActions } from "./middlewares/group-specific-actions" import { MentionListener } from "./middlewares/mention-listener" import { MessageLink } from "./middlewares/message-link" @@ -83,6 +84,7 @@ bot.use(async (ctx, next) => { bot.use(MenuGenerator.getInstance()) bot.use(commands) bot.use(new BotMembershipHandler()) +bot.use(new GroupAdminsSync()) bot.use(new CampaignSpamGuard()) bot.use(new AutoModerationStack()) bot.use(new GroupSpecificActions()) diff --git a/src/lib/group-admins/index.ts b/src/lib/group-admins/index.ts new file mode 100644 index 0000000..79c41d8 --- /dev/null +++ b/src/lib/group-admins/index.ts @@ -0,0 +1,102 @@ +import type { ChatMember } from "grammy/types" + +type Logger = { + warn: (obj: object, msg: string) => void + info: (obj: object, msg: string) => void +} + +/** Where the bot keeps what Telegram last said about each group's admins: a cache, never a source. */ +export type GroupAdminsStore = { + read: (chatId: number) => Promise + write: (chatId: number, userIds: number[]) => Promise +} + +export type GroupAdminsSyncDeps = { + /** Telegram's `getChatAdministrators`. */ + getChatAdministrators: (chatId: number) => Promise + /** The Telegram IDs of the groups the backend stores. */ + listGroups: () => Promise + store: GroupAdminsStore + logger: Logger + /** Pause between Telegram requests during a full refresh. */ + wait: (ms: number) => Promise + /** How long to wait for more admin changes in a chat before reading it once. */ + debounceMs?: number + requestIntervalMs?: number +} + +const isAdminStatus = (status: ChatMember["status"]) => status === "creator" || status === "administrator" + +/** The people who administer a chat: its creator and administrators, without bots. */ +export function toAdminIds(members: ChatMember[]): number[] { + return [...new Set(members.filter((m) => isAdminStatus(m.status) && !m.user.is_bot).map((m) => m.user.id))] +} + +/** Whether a `chat_member` update changes who administers the chat. */ +export function changesAdmins(oldMember: ChatMember, newMember: ChatMember): boolean { + if (newMember.user.is_bot) return false + return oldMember.status !== newMember.status && (isAdminStatus(oldMember.status) || isAdminStatus(newMember.status)) +} + +/** + * Keeps the bot's cache of each group's native Telegram admins (RFC v3 decision 2) fresh. Telegram is the only source: + * the backend asks the bot for the cached lists and stores nothing. + */ +export function createGroupAdminsSync(deps: GroupAdminsSyncDeps) { + const debounceMs = deps.debounceMs ?? 5_000 + const requestIntervalMs = deps.requestIntervalMs ?? 100 + const pending = new Map>() + let fullRefreshRunning = false + + async function refreshChat(chatId: number): Promise { + try { + await deps.store.write(chatId, toAdminIds(await deps.getChatAdministrators(chatId))) + return true + } catch (error) { + deps.logger.warn({ error, chatId }, "[GroupAdmins] cannot read the chat's admins") + return false + } + } + + return { + refreshChat, + + /** Reads a chat once its admin changes settle, so a burst of promotions makes one request. */ + schedule(chatId: number) { + clearTimeout(pending.get(chatId)) + pending.set( + chatId, + setTimeout(() => { + pending.delete(chatId) + void refreshChat(chatId) + }, debounceMs) + ) + }, + + /** Reads every stored group, one Telegram request at a time. Skipped while another full refresh runs. */ + async refreshAll(): Promise<{ refreshed: number; total: number } | null> { + if (fullRefreshRunning) return null + fullRefreshRunning = true + try { + const groups = await deps.listGroups() + let refreshed = 0 + for (const [index, chatId] of groups.entries()) { + if (index > 0) await deps.wait(requestIntervalMs) + if (await refreshChat(chatId)) refreshed++ + } + deps.logger.info({ refreshed, total: groups.length }, "[GroupAdmins] full refresh done") + return { refreshed, total: groups.length } + } finally { + fullRefreshRunning = false + } + }, + + /** The cached admins of the given groups, for the backend; groups not read yet are left out. */ + async cached(chatIds: number[]): Promise<{ chatId: number; userIds: number[] }[]> { + const entries = await Promise.all( + chatIds.map(async (chatId) => ({ chatId, userIds: await deps.store.read(chatId).catch(() => undefined) })) + ) + return entries.flatMap(({ chatId, userIds }) => (userIds ? [{ chatId, userIds }] : [])) + }, + } +} diff --git a/src/middlewares/group-admins-sync.ts b/src/middlewares/group-admins-sync.ts new file mode 100644 index 0000000..9032873 --- /dev/null +++ b/src/middlewares/group-admins-sync.ts @@ -0,0 +1,55 @@ +import { Cron } from "croner" +import type { Context } from "grammy" +import { serviceBackend } from "@/backend" +import { changesAdmins, createGroupAdminsSync } from "@/lib/group-admins" +import { RedisFallbackAdapter } from "@/lib/redis-fallback-adapter" +import { logger } from "@/logger" +import { modules } from "@/modules" +import { type TelemetryContextFlavor, TrackedMiddleware } from "@/modules/telemetry" +import { redis } from "@/redis" +import { wait } from "@/utils/wait" + +/** After joining, give BotMembershipHandler time to store the group. */ +const JOIN_DELAY_MS = 30_000 +const STARTUP_DELAY_MS = 120_000 +/** Refreshed every 6 hours: an entry that outlives a day belongs to a group the bot can no longer read. */ +const CACHE_TTL_SECONDS = 24 * 60 * 60 + +const cache = new RedisFallbackAdapter({ redis, prefix: "group_admins", ttl: CACHE_TTL_SECONDS, logger }) + +/** The bot's cache of each group's native Telegram admins (RFC v3 decision 2), which the backend reads for People. */ +export const groupAdmins = createGroupAdminsSync({ + getChatAdministrators: (chatId) => modules.shared.api.getChatAdministrators(chatId), + listGroups: async () => (await serviceBackend.tg.groups.getAll.query()).map((group) => group.telegramId), + store: { + read: (chatId) => cache.read(String(chatId)), + write: (chatId, userIds) => cache.write(String(chatId), userIds), + }, + logger, + wait, +}) + +/** Refreshes the cache on admin changes, when the bot is promoted in a group, at startup and every 6 hours. */ +export class GroupAdminsSync> extends TrackedMiddleware { + constructor() { + super("group_admins_sync") + + const refreshAll = () => + void groupAdmins.refreshAll().catch((error) => logger.error({ error }, "[GroupAdmins] full refresh failed")) + setTimeout(refreshAll, STARTUP_DELAY_MS) + new Cron("0 17 */6 * * *", refreshAll) + + this.composer.on("chat_member", async (ctx, next) => { + const { chat, old_chat_member: oldMember, new_chat_member: newMember } = ctx.chatMember + if (chat.type !== "private" && changesAdmins(oldMember, newMember)) groupAdmins.schedule(chat.id) + await next() + }) + + this.composer.on("my_chat_member", async (ctx, next) => { + const { chat, new_chat_member: me } = ctx.myChatMember + if (chat.type !== "private" && me.status === "administrator") + setTimeout(() => void groupAdmins.refreshChat(chat.id), JOIN_DELAY_MS) + await next() + }) + } +} diff --git a/src/modules/websocket.ts b/src/modules/websocket.ts index c62a885..52b47ba 100644 --- a/src/modules/websocket.ts +++ b/src/modules/websocket.ts @@ -165,6 +165,11 @@ export class WebSocketClient extends Module { + const { groupAdmins } = await import("@/middlewares/group-admins-sync") + cb(await groupAdmins.cached(chatIds).catch(() => [])) + }) + this.io.on("leaveChat", async ({ chatId, performer }: { chatId: number; performer: SocketActor }, cb) => { const ok = await this.shared.api.leaveChat(chatId).catch(() => false) diff --git a/tests/group-admins.test.ts b/tests/group-admins.test.ts new file mode 100644 index 0000000..70866f6 --- /dev/null +++ b/tests/group-admins.test.ts @@ -0,0 +1,115 @@ +import type { ChatMember } from "grammy/types" +import { afterEach, describe, expect, it, vi } from "vitest" +import { changesAdmins, createGroupAdminsSync, type GroupAdminsSyncDeps, toAdminIds } from "@/lib/group-admins" + +const member = (id: number, status: ChatMember["status"], is_bot = false) => + ({ status, user: { id, is_bot, first_name: `User ${id}` } }) as ChatMember + +function deps(overrides: Partial = {}) { + const cache = new Map() + return { + cache, + getChatAdministrators: vi.fn(async () => [member(1, "creator"), member(2, "administrator")]), + listGroups: vi.fn(async () => [-1001, -1002]), + store: { + read: vi.fn(async (chatId: number) => cache.get(chatId)), + write: vi.fn(async (chatId: number, userIds: number[]) => void cache.set(chatId, userIds)), + }, + logger: { warn: vi.fn(), info: vi.fn() }, + wait: vi.fn(async () => {}), + ...overrides, + } satisfies GroupAdminsSyncDeps & { cache: Map } +} + +afterEach(() => { + vi.useRealTimers() +}) + +describe("toAdminIds", () => { + it("keeps the creator and administrators, without bots", () => { + expect( + toAdminIds([ + member(1, "creator"), + member(2, "administrator"), + member(3, "administrator", true), + member(4, "member"), + ]) + ).toEqual([1, 2]) + }) +}) + +describe("changesAdmins", () => { + it("is true only when someone becomes or stops being an admin", () => { + expect(changesAdmins(member(1, "member"), member(1, "administrator"))).toBe(true) + expect(changesAdmins(member(1, "administrator"), member(1, "left"))).toBe(true) + expect(changesAdmins(member(1, "administrator"), member(1, "creator"))).toBe(true) + expect(changesAdmins(member(1, "member"), member(1, "restricted"))).toBe(false) + expect(changesAdmins(member(1, "administrator"), member(1, "administrator"))).toBe(false) + expect(changesAdmins(member(9, "member"), member(9, "administrator", true))).toBe(false) + }) +}) + +describe("createGroupAdminsSync", () => { + it("caches what Telegram says about a chat's admins", async () => { + const d = deps() + expect(await createGroupAdminsSync(d).refreshChat(-1001)).toBe(true) + expect(d.cache.get(-1001)).toEqual([1, 2]) + }) + + it("keeps the previous entry when Telegram cannot list the admins", async () => { + const d = deps({ getChatAdministrators: vi.fn(async () => Promise.reject(new Error("chat not found"))) }) + d.cache.set(-1001, [7]) + expect(await createGroupAdminsSync(d).refreshChat(-1001)).toBe(false) + expect(d.cache.get(-1001)).toEqual([7]) + expect(d.logger.warn).toHaveBeenCalled() + }) + + it("coalesces a burst of admin changes in a chat into one read", async () => { + vi.useFakeTimers() + const d = deps() + const sync = createGroupAdminsSync({ ...d, debounceMs: 1000 }) + sync.schedule(-1001) + sync.schedule(-1001) + await vi.advanceTimersByTimeAsync(500) + sync.schedule(-1001) + await vi.advanceTimersByTimeAsync(1000) + expect(d.getChatAdministrators).toHaveBeenCalledTimes(1) + }) + + it("refreshes every group and keeps going after a failure", async () => { + const d = deps({ + listGroups: vi.fn(async () => [-1001, -1002, -1003]), + getChatAdministrators: vi + .fn() + .mockResolvedValueOnce([member(1, "creator")]) + .mockRejectedValueOnce(new Error("kicked")) + .mockResolvedValueOnce([]), + }) + expect(await createGroupAdminsSync(d).refreshAll()).toEqual({ refreshed: 2, total: 3 }) + expect([...d.cache.entries()]).toEqual([ + [-1001, [1]], + [-1003, []], + ]) + expect(d.wait).toHaveBeenCalledTimes(2) + }) + + it("skips a full refresh while another one runs", async () => { + let release = () => {} + const d = deps({ listGroups: vi.fn(() => new Promise((resolve) => (release = () => resolve([])))) }) + const sync = createGroupAdminsSync(d) + const first = sync.refreshAll() + expect(await sync.refreshAll()).toBeNull() + release() + expect(await first).toEqual({ refreshed: 0, total: 0 }) + }) + + it("answers the backend with the cached groups only, empty lists included", async () => { + const d = deps() + d.cache.set(-1001, [1, 2]) + d.cache.set(-1002, []) + expect(await createGroupAdminsSync(d).cached([-1001, -1002, -1003])).toEqual([ + { chatId: -1001, userIds: [1, 2] }, + { chatId: -1002, userIds: [] }, + ]) + }) +})