feat(groups): answer the backend with each group's native admins

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())

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<number[] | undefined>
+  write: (chatId: number, userIds: number[]) => Promise<void>
+}
+
+export type GroupAdminsSyncDeps = {
+  /** Telegram's `getChatAdministrators`. */
+  getChatAdministrators: (chatId: number) => Promise<ChatMember[]>
+  /** The Telegram IDs of the groups the backend stores. */
+  listGroups: () => Promise<number[]>
+  store: GroupAdminsStore
+  logger: Logger
+  /** Pause between Telegram requests during a full refresh. */
+  wait: (ms: number) => Promise<void>
+  /** 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<number, ReturnType<typeof setTimeout>>()
+  let fullRefreshRunning = false
+
+  async function refreshChat(chatId: number): Promise<boolean> {
+    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 }] : []))
+    },
+  }
+}

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<number[]>({ 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<C extends TelemetryContextFlavor<Context>> extends TrackedMiddleware<C> {
+  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()
+    })
+  }
+}

src/modules/websocket.ts

@@ -165,6 +165,11 @@
       cb(null)
     })
 
+    this.io.on("getGroupAdmins", async ({ chatIds }: { chatIds: number[] }, cb) => {
+      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)
 

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<GroupAdminsSyncDeps> = {}) {
+  const cache = new Map<number, number[]>()
+  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<number, number[]> }
+}
+
+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<GroupAdminsSyncDeps["getChatAdministrators"]>()
+        .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<number[]>((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: [] },
+    ])
+  })
+})