feat(retention): delete expired sessions and ended moderation records

src/constants.ts

@@ -12,6 +12,12 @@ export const TRUSTED_ORIGINS = [
 ]
 
 export const MESSAGES_RETENTION_DAYS = 7
+/** Expired sessions are deleted this many days after they expire (privacy policy, data retention). */
+export const SESSIONS_RETENTION_DAYS = 30
+/** Temporary moderation measures are deleted this many months after they end (privacy policy, data retention). */
+export const MODERATION_RETENTION_MONTHS = 24
+/** How long revocations of permanent measures stay in the moderation audit log; see `endedMeasures`. */
+export const ENDED_MODERATION_RETENTION_DAYS = 1
 
 export const USER_ROLE = {
   ADMIN: "admin",

src/cron.ts

@@ -1,15 +1,18 @@
 import { Cron } from "croner"
 import { lt } from "drizzle-orm"
-import { MESSAGES_RETENTION_DAYS } from "./constants"
+import { MESSAGES_RETENTION_DAYS, SESSIONS_RETENTION_DAYS } from "./constants"
 import { DB, SCHEMA } from "./db"
 import { env } from "./env"
 import { logger } from "./logger"
+import { endedMeasures } from "./utils/audit-log"
 
 export function cron() {
   new Cron("0 23 4 * * *", async () => {
     logger.info("[CRON] start")
     await cleanMessages()
     await cleanLinkCodes()
+    await cleanSessions()
+    await cleanAuditLog()
     logger.info("[CRON] end")
   })
 
@@ -63,3 +66,31 @@ async function cleanMessages() {
     logger.error(err, "[CRON] cleanMessages")
   }
 }
+
+export async function cleanSessions() {
+  logger.info(`[CRON] START(cleanSessions) retention days = ${SESSIONS_RETENTION_DAYS}`)
+
+  const expiredBefore = new Date()
+  expiredBefore.setDate(expiredBefore.getDate() - SESSIONS_RETENTION_DAYS)
+  try {
+    const deleted = await DB.delete(SCHEMA.AUTH.session)
+      .where(lt(SCHEMA.AUTH.session.expiresAt, expiredBefore))
+      .returning({ id: SCHEMA.AUTH.session.id })
+    logger.info(`[CRON] END(cleanSessions) deleted ${deleted.length} session(s).`)
+  } catch (err) {
+    logger.error(err, "[CRON] cleanSessions")
+  }
+}
+
+export async function cleanAuditLog() {
+  logger.info(`[CRON] START(cleanAuditLog)`)
+
+  try {
+    const deleted = await DB.delete(SCHEMA.TG.auditLog)
+      .where(endedMeasures(new Date()))
+      .returning({ id: SCHEMA.TG.auditLog.id })
+    logger.info(`[CRON] END(cleanAuditLog) deleted ${deleted.length} audit log record(s).`)
+  } catch (err) {
+    logger.error(err, "[CRON] cleanAuditLog")
+  }
+}

src/routers/tg/audit-log.ts

@@ -12,6 +12,7 @@ import {
 } from "@/db/schema/tg/audit-log"
 import { dashboard, SCOPE } from "@/idp/policies"
 import { createTRPCRouter, policy } from "@/trpc"
+import { revokedMeasures } from "@/utils/audit-log"
 import { decryptUser } from "@/utils/users"
 
 const AUDIT_LOG = SCHEMA.TG.auditLog
@@ -81,23 +82,31 @@ export default createTRPCRouter({
         else if (!backed) review = AUDIT_REVIEW.PERMISSION_MISSING
       }
 
-      const [inserted] = await DB.insert(AUDIT_LOG)
-        .values({
-          adminId: input.actorTelegramId,
-          actorTgId: input.actorTelegramId,
-          actorSub: subject?.sub ?? null,
-          targetId: input.targetId,
-          groupId: input.groupId,
-          type: input.type,
-          until: input.until,
-          reason: input.reason,
-          client: actor.client,
-          basis: input.basis,
-          review,
-          idempotencyKey: input.idempotencyKey,
-        })
-        .onConflictDoNothing({ target: AUDIT_LOG.idempotencyKey })
-        .returning({ id: AUDIT_LOG.id })
+      // A revocation deletes the records of the measures it lifts, in the same transaction so a
+      // failed delete is retried with the record instead of being skipped as a duplicate.
+      const inserted = await DB.transaction(async (tx) => {
+        const [row] = await tx
+          .insert(AUDIT_LOG)
+          .values({
+            adminId: input.actorTelegramId,
+            actorTgId: input.actorTelegramId,
+            actorSub: subject?.sub ?? null,
+            targetId: input.targetId,
+            groupId: input.groupId,
+            type: input.type,
+            until: input.until,
+            reason: input.reason,
+            client: actor.client,
+            basis: input.basis,
+            review,
+            idempotencyKey: input.idempotencyKey,
+          })
+          .onConflictDoNothing({ target: AUDIT_LOG.idempotencyKey })
+          .returning({ id: AUDIT_LOG.id })
+        const revoked = row && revokedMeasures({ ...input, id: row.id })
+        if (revoked) await tx.delete(AUDIT_LOG).where(revoked)
+        return row
+      })
       if (inserted) return { id: inserted.id, duplicate: false }
 
       const [existing] = await DB.select({ id: AUDIT_LOG.id })

src/utils/audit-log.ts

@@ -0,0 +1,108 @@
+import { and, eq, inArray, isNotNull, isNull, lt, notExists, or, type SQL, sql } from "drizzle-orm"
+import { type AnyPgColumn, alias, QueryBuilder } from "drizzle-orm/pg-core"
+import { ENDED_MODERATION_RETENTION_DAYS, MODERATION_RETENTION_MONTHS } from "@/constants"
+import { AUDIT_TYPE, auditLog, type TAuditType } from "@/db/schema/tg/audit-log"
+
+// Moderation records are kept for 24 months after expiry if temporary, until revocation if
+// permanent (privacy policy, data retention). A kick is a temporary measure that ends at once.
+
+/** The measures each revocation lifts; `allGroups` when it lifts them in every group. */
+const REVOKES: Partial<Record<TAuditType, { types: TAuditType[]; allGroups: boolean }>> = {
+  [AUDIT_TYPE.UNBAN]: { types: [AUDIT_TYPE.BAN], allGroups: false },
+  [AUDIT_TYPE.UNMUTE]: { types: [AUDIT_TYPE.MUTE], allGroups: false },
+  // The bot's unban_all unbans the target in every group, which also lifts single-group bans.
+  [AUDIT_TYPE.UNBAN_ALL]: { types: [AUDIT_TYPE.BAN_ALL, AUDIT_TYPE.BAN], allGroups: true },
+}
+
+const MEASURES: TAuditType[] = [AUDIT_TYPE.BAN, AUDIT_TYPE.MUTE, AUDIT_TYPE.BAN_ALL]
+const REVOCATIONS: TAuditType[] = [AUDIT_TYPE.UNBAN, AUDIT_TYPE.UNMUTE, AUDIT_TYPE.UNBAN_ALL]
+
+/** Telegram treats a restriction ending more than 366 days after it is applied as permanent. */
+const TELEGRAM_FOREVER_AFTER = sql.raw(`interval '366 days'`)
+
+type MeasureRecord = { until: AnyPgColumn; createdAt: AnyPgColumn }
+
+/** A measure with an end date Telegram honours. */
+function temporary(record: MeasureRecord) {
+  return and(isNotNull(record.until), sql`${record.until} <= ${record.createdAt} + ${TELEGRAM_FOREVER_AFTER}`)
+}
+
+/** A measure without an end date, or with one Telegram ignores. */
+function permanent(record: MeasureRecord) {
+  return or(isNull(record.until), sql`${record.until} > ${record.createdAt} + ${TELEGRAM_FOREVER_AFTER}`)
+}
+
+/**
+ * The records `revocation` deletes, or undefined when it is not a revocation: the permanent
+ * measures it lifts. Lifted temporary measures stay until their retention ends. Only records
+ * stored before the revocation match, so it never lifts a later measure.
+ */
+export function revokedMeasures(revocation: {
+  id: number
+  type: TAuditType
+  targetId: number
+  groupId: number | null
+}): SQL | undefined {
+  const revokes = REVOKES[revocation.type]
+  if (!revokes) return undefined
+
+  const group = revokes.allGroups
+    ? undefined
+    : revocation.groupId === null
+      ? isNull(auditLog.groupId)
+      : eq(auditLog.groupId, revocation.groupId)
+  return and(
+    eq(auditLog.targetId, revocation.targetId),
+    inArray(auditLog.type, revokes.types),
+    lt(auditLog.id, revocation.id),
+    group,
+    permanent(auditLog)
+  )
+}
+
+/** Whether the revocation in the outer `auditLog` row lifted a temporary measure still in force. */
+function liftsTemporaryMeasure() {
+  const lifted = alias(auditLog, "lifted")
+  const revokes = Object.entries(REVOKES).map(([type, { types, allGroups }]) =>
+    and(
+      eq(auditLog.type, type as TAuditType),
+      inArray(lifted.type, types),
+      allGroups ? undefined : sql`${lifted.groupId} is not distinct from ${auditLog.groupId}`
+    )
+  )
+  return new QueryBuilder()
+    .select({ one: sql`1` })
+    .from(lifted)
+    .where(
+      and(
+        eq(lifted.targetId, auditLog.targetId),
+        lt(lifted.id, auditLog.id),
+        or(...revokes),
+        temporary(lifted),
+        sql`${lifted.until} > ${auditLog.createdAt}`
+      )
+    )
+}
+
+/**
+ * Records past their retention at `now`:
+ * - temporary measures {@link MODERATION_RETENTION_MONTHS} after they end, kicks the same time
+ *   after they are made;
+ * - revocations of a temporary measure as long as a temporary measure, from when they are made;
+ * - other revocations after {@link ENDED_MODERATION_RETENTION_DAYS}, long enough for the bot's
+ *   retries of a revocation to hit its idempotency key instead of lifting a later measure.
+ *
+ * Permanent measures are deleted only by their revocation, see {@link revokedMeasures}.
+ */
+export function endedMeasures(now: Date): SQL | undefined {
+  const retainedSince = new Date(now)
+  retainedSince.setUTCMonth(retainedSince.getUTCMonth() - MODERATION_RETENTION_MONTHS)
+  const retriesSince = new Date(now)
+  retriesSince.setDate(retriesSince.getDate() - ENDED_MODERATION_RETENTION_DAYS)
+
+  return or(
+    and(inArray(auditLog.type, MEASURES), temporary(auditLog), lt(auditLog.until, retainedSince)),
+    and(inArray(auditLog.type, [AUDIT_TYPE.KICK, ...REVOCATIONS]), lt(auditLog.createdAt, retainedSince)),
+    and(inArray(auditLog.type, REVOCATIONS), lt(auditLog.createdAt, retriesSince), notExists(liftsTemporaryMeasure()))
+  )
+}

tests/audit-log-retention.test.ts

@@ -0,0 +1,97 @@
+import { PgDialect } from "drizzle-orm/pg-core"
+import { describe, expect, it } from "vitest"
+import { endedMeasures, revokedMeasures } from "@/utils/audit-log"
+
+const dialect = new PgDialect()
+function render(condition: Parameters<PgDialect["sqlToQuery"]>[0] | undefined) {
+  if (!condition) return undefined
+  const { sql, params } = dialect.sqlToQuery(condition)
+  return { sql, params }
+}
+
+/** No end date, or one Telegram treats as forever. */
+const PERMANENT =
+  '("tg_audit_log"."until" is null or "tg_audit_log"."until" > "tg_audit_log"."created_at" + interval \'366 days\')'
+/** An end date Telegram honours. */
+const TEMPORARY = (table: string) =>
+  `("${table}"."until" is not null and "${table}"."until" <= "${table}"."created_at" + interval '366 days')`
+
+describe("revokedMeasures", () => {
+  it("matches the earlier permanent bans of the target in the same group on unban", () => {
+    expect(render(revokedMeasures({ id: 9, type: "unban", targetId: 777, groupId: -100 }))).toEqual({
+      sql: `("tg_audit_log"."target_id" = $1 and "tg_audit_log"."type" in ($2) and "tg_audit_log"."id" < $3 and "tg_audit_log"."group_id" = $4 and ${PERMANENT})`,
+      params: [777, "ban", 9, -100],
+    })
+  })
+
+  it("matches the earlier mutes of the target in the same group on unmute", () => {
+    expect(render(revokedMeasures({ id: 9, type: "unmute", targetId: 777, groupId: -100 }))?.params).toEqual([
+      777,
+      "mute",
+      9,
+      -100,
+    ])
+  })
+
+  it("matches the permanent ban_all and bans in every group on unban_all", () => {
+    expect(render(revokedMeasures({ id: 9, type: "unban_all", targetId: 777, groupId: null }))).toEqual({
+      sql: `("tg_audit_log"."target_id" = $1 and "tg_audit_log"."type" in ($2, $3) and "tg_audit_log"."id" < $4 and ${PERMANENT})`,
+      params: [777, "ban_all", "ban", 9],
+    })
+  })
+
+  it("matches nothing for a measure", () => {
+    for (const type of ["ban", "mute", "kick", "ban_all"] as const) {
+      expect(revokedMeasures({ id: 9, type, targetId: 777, groupId: -100 })).toBeUndefined()
+    }
+  })
+})
+
+describe("endedMeasures", () => {
+  const now = new Date("2026-10-14T12:00:00Z")
+  const twoYearsAgo = new Date("2024-10-14T12:00:00Z").toISOString()
+  const yesterday = new Date("2026-10-13T12:00:00Z").toISOString()
+  const ended = render(endedMeasures(now))
+
+  it("matches temporary measures 24 months after they end", () => {
+    expect(ended?.sql).toContain(
+      `("tg_audit_log"."type" in ($1, $2, $3) and ${TEMPORARY("tg_audit_log")} and "tg_audit_log"."until" < $4)`
+    )
+    expect(ended?.params.slice(0, 4)).toEqual(["ban", "mute", "ban_all", twoYearsAgo])
+  })
+
+  it("matches kicks and revocations 24 months after they are made", () => {
+    expect(ended?.sql).toContain('("tg_audit_log"."type" in ($5, $6, $7, $8) and "tg_audit_log"."created_at" < $9)')
+    expect(ended?.params.slice(4, 9)).toEqual(["kick", "unban", "unmute", "unban_all", twoYearsAgo])
+  })
+
+  it("matches revocations older than a day that lifted no temporary measure in force", () => {
+    expect(ended?.sql).toContain(
+      '("tg_audit_log"."type" in ($10, $11, $12) and "tg_audit_log"."created_at" < $13 and not exists (select 1 from "tg_audit_log" "lifted" where ('
+    )
+    expect(ended?.sql).toContain(
+      '"lifted"."target_id" = "tg_audit_log"."target_id" and "lifted"."id" < "tg_audit_log"."id" and ' +
+        '(("tg_audit_log"."type" = $14 and "lifted"."type" in ($15) and "lifted"."group_id" is not distinct from "tg_audit_log"."group_id") or ' +
+        '("tg_audit_log"."type" = $16 and "lifted"."type" in ($17) and "lifted"."group_id" is not distinct from "tg_audit_log"."group_id") or ' +
+        '("tg_audit_log"."type" = $18 and "lifted"."type" in ($19, $20))) and ' +
+        `${TEMPORARY("lifted")} and "lifted"."until" > "tg_audit_log"."created_at"`
+    )
+    expect(ended?.params.slice(9)).toEqual([
+      "unban",
+      "unmute",
+      "unban_all",
+      yesterday,
+      "unban",
+      "ban",
+      "unmute",
+      "mute",
+      "unban_all",
+      "ban_all",
+      "ban",
+    ])
+  })
+
+  it("never matches a permanent measure", () => {
+    expect(ended?.sql).not.toContain('"tg_audit_log"."until" is null')
+  })
+})

tests/cron-link-cleanup.test.ts

@@ -20,6 +20,7 @@ vi.mock("@/db", () => ({
 }))
 vi.mock("@/env", () => ({ env: { NODE_ENV: "test" } }))
 vi.mock("@/logger", () => ({ logger: { error: vi.fn(), info: vi.fn() } }))
+vi.mock("@/utils/audit-log", () => ({ endedMeasures: vi.fn() }))
 
 const { cleanLinkCodes } = await import("@/cron")
 

tests/cron-retention.test.ts

@@ -0,0 +1,61 @@
+import { beforeEach, describe, expect, it, vi } from "vitest"
+
+const state = vi.hoisted(() => {
+  const session = { id: Symbol("auth_sessions.id"), expiresAt: Symbol("auth_sessions.expires_at") }
+  const auditLog = { id: Symbol("tg_audit_log.id") }
+  const condition = Symbol("expiry-condition")
+  const ended = Symbol("ended-measures")
+  const returning = vi.fn(async () => [{ id: 1 }])
+  const where = vi.fn(() => ({ returning }))
+  const deleteFrom = vi.fn(() => ({ where }))
+  const lessThan = vi.fn(() => condition)
+  const endedMeasures = vi.fn(() => ended)
+
+  return { auditLog, condition, deleteFrom, ended, endedMeasures, lessThan, returning, session, where }
+})
+
+vi.mock("croner", () => ({ Cron: vi.fn() }))
+vi.mock("drizzle-orm", () => ({ lt: state.lessThan }))
+vi.mock("@/db", () => ({
+  DB: { delete: state.deleteFrom },
+  SCHEMA: { AUTH: { session: state.session }, TG: { auditLog: state.auditLog } },
+}))
+vi.mock("@/env", () => ({ env: { NODE_ENV: "test" } }))
+vi.mock("@/logger", () => ({ logger: { error: vi.fn(), info: vi.fn() } }))
+vi.mock("@/utils/audit-log", () => ({ endedMeasures: state.endedMeasures }))
+
+const { cleanAuditLog, cleanSessions } = await import("@/cron")
+
+const DAY = 86_400_000
+
+beforeEach(() => {
+  vi.clearAllMocks()
+})
+
+describe("cleanSessions", () => {
+  it("deletes sessions that expired more than 30 days ago", async () => {
+    const before = Date.now()
+    await cleanSessions()
+
+    expect(state.deleteFrom).toHaveBeenCalledWith(state.session)
+    expect(state.lessThan).toHaveBeenCalledWith(state.session.expiresAt, expect.any(Date))
+    const cutoff = (state.lessThan.mock.calls[0] as unknown[])[1] as Date
+    // setDate keeps the wall-clock time, so a DST change can shift the cutoff by an hour.
+    expect(Math.abs(before - cutoff.getTime() - 30 * DAY)).toBeLessThanOrEqual(3_600_000 + 1_000)
+    expect(state.where).toHaveBeenCalledWith(state.condition)
+  })
+})
+
+describe("cleanAuditLog", () => {
+  it("deletes the moderation records past their retention", async () => {
+    const before = Date.now()
+    await cleanAuditLog()
+
+    expect(state.endedMeasures).toHaveBeenCalledWith(expect.any(Date))
+    const now = (state.endedMeasures.mock.calls[0] as unknown[])[0] as Date
+    expect(now.getTime()).toBeGreaterThanOrEqual(before)
+    expect(now.getTime()).toBeLessThanOrEqual(Date.now())
+    expect(state.deleteFrom).toHaveBeenCalledWith(state.auditLog)
+    expect(state.where).toHaveBeenCalledWith(state.ended)
+  })
+})