feat(federation): add soft-delete GC for empty group DMs with 24h grace period

Replace the hard-delete in the leave handler with a soft-delete (sets
deleted_at timestamp) when the last member leaves a group DM. A new
janitor sweep in the federation worker runs hourly and purges channels
whose grace period has expired, cascading through reactions, embeds,
attachments, messages, members, outbox/mutation-log/file-queue entries,
and finally the channel itself.

All client-facing dm_channels queries now filter on deleted_at IS NULL
to hide soft-deleted channels from the REST API and WebSocket ready
payload.
This commit is contained in:
Jannis Braun
2026-03-26 20:35:55 +01:00
parent 62a16e884a
commit 4efa35f311
4 changed files with 177 additions and 45 deletions
+15 -40
View File
@@ -1,5 +1,5 @@
import type { FastifyInstance } from 'fastify';
import { eq, and, or, desc, lt, inArray, sql } from 'drizzle-orm';
import { eq, and, or, desc, lt, inArray, isNull, sql } from 'drizzle-orm';
import { getDb, schema } from '../db/index.js';
import { authenticate } from '../utils/auth.js';
import { generateSnowflake } from '../utils/snowflake.js';
@@ -19,7 +19,7 @@ import {
type Embed,
} from '@backspace/shared';
import { sanitizeUser } from '../utils/sanitize.js';
import { deleteUploadFile, deleteAttachmentFiles } from '../utils/fileCleanup.js';
import { deleteAttachmentFiles } from '../utils/fileCleanup.js';
import { fetchDmEmbedsForMessages, resolveEmbeds, reResolveEmbeds, embedRowToEmbed } from '../utils/embedResolver.js';
import {
appendMutationLog,
@@ -198,7 +198,7 @@ export function broadcastDmMessage(dmChannelId: string, message: DmMessageWithUs
const dmChannel = db.select()
.from(schema.dmChannels)
.where(eq(schema.dmChannels.id, dmChannelId))
.where(and(eq(schema.dmChannels.id, dmChannelId), isNull(schema.dmChannels.deletedAt)))
.get();
if (dmChannel) {
@@ -243,9 +243,9 @@ export async function dmRoutes(app: FastifyInstance): Promise<void> {
const dmChannelIds = memberships.map(m => m.dmChannelId);
// Batch fetch all DM channels
// Batch fetch all DM channels (exclude soft-deleted)
const channelRows = db.select().from(schema.dmChannels)
.where(inArray(schema.dmChannels.id, dmChannelIds)).all();
.where(and(inArray(schema.dmChannels.id, dmChannelIds), isNull(schema.dmChannels.deletedAt))).all();
const channelMap = new Map(channelRows.map(c => [c.id, c]));
// Batch fetch all DM members
@@ -383,10 +383,10 @@ export async function dmRoutes(app: FastifyInstance): Promise<void> {
.length;
if (memberCount !== 2) continue;
// DM channel already exists between these users
// DM channel already exists between these users (exclude soft-deleted)
const dmChannel = db.select()
.from(schema.dmChannels)
.where(eq(schema.dmChannels.id, myDm.dmChannelId))
.where(and(eq(schema.dmChannels.id, myDm.dmChannelId), isNull(schema.dmChannels.deletedAt)))
.get();
if (!dmChannel) continue;
@@ -538,7 +538,7 @@ export async function dmRoutes(app: FastifyInstance): Promise<void> {
}
// Enforce DM channel ownership: only the owner can add members (for new-style group DMs)
let dmChannel = db.select().from(schema.dmChannels).where(eq(schema.dmChannels.id, id)).get();
let dmChannel = db.select().from(schema.dmChannels).where(and(eq(schema.dmChannels.id, id), isNull(schema.dmChannels.deletedAt))).get();
if (!dmChannel) {
return reply.code(404).send({ error: 'DM channel not found', statusCode: 404 });
}
@@ -795,7 +795,7 @@ export async function dmRoutes(app: FastifyInstance): Promise<void> {
}
// Check DM channel ownership before leaving
const dmChannel = db.select().from(schema.dmChannels).where(eq(schema.dmChannels.id, id)).get();
const dmChannel = db.select().from(schema.dmChannels).where(and(eq(schema.dmChannels.id, id), isNull(schema.dmChannels.deletedAt))).get();
// Compute federation targets BEFORE member deletion so the leaving user's peer is included
let fedTargetOrigins: string[] | undefined;
@@ -932,37 +932,12 @@ export async function dmRoutes(app: FastifyInstance): Promise<void> {
});
}
} else {
// Last member left — clean up the entire DM channel
// Collect attachment filenames for disk cleanup
const msgIds = db.select({ id: schema.dmMessages.id })
.from(schema.dmMessages)
.where(eq(schema.dmMessages.dmChannelId, id))
.all()
.map(m => m.id);
const filesToDelete: { filename: string }[] = [];
if (msgIds.length > 0) {
const attachmentRows = db.select({ filename: schema.attachments.filename })
.from(schema.attachments)
.where(inArray(schema.attachments.dmMessageId, msgIds))
.all();
filesToDelete.push(...attachmentRows);
// Delete attachments and reactions before cascade
db.transaction((tx) => {
tx.delete(schema.attachments).where(inArray(schema.attachments.dmMessageId, msgIds)).run();
tx.delete(schema.dmReactions).where(inArray(schema.dmReactions.dmMessageId, msgIds)).run();
});
}
// Clean up all read_states for this DM channel (all members' rows)
db.delete(schema.readStates).where(eq(schema.readStates.channelId, id)).run();
// Delete the DM channel (cascades to dm_messages)
db.delete(schema.dmChannels).where(eq(schema.dmChannels.id, id)).run();
// Clean up files from disk
deleteAttachmentFiles(filesToDelete);
// Last member left — soft-delete for deferred GC (24h grace period)
db.update(schema.dmChannels)
.set({ deletedAt: Date.now() })
.where(eq(schema.dmChannels.id, id))
.run();
console.log(`[dm] Group DM ${id} has no remaining members, soft-deleted for GC`);
}
// Send dm_channel_closed to the leaving user
@@ -3,6 +3,7 @@ import * as schema from '../db/schema.js';
import { eq, and, lte, asc, inArray } from 'drizzle-orm';
import { config } from '../config.js';
import { isFederationRelayEnabled } from './federationOutbox.js';
import { runFederationJanitor } from './storageJanitor.js';
import { buildFederationHeaders, getOurOrigin } from './federationAuth.js';
import { generateSnowflake } from './snowflake.js';
import { getDmMessageWithUser } from '../routes/dm.js';
@@ -20,6 +21,7 @@ import { Readable } from 'node:stream';
const OUTBOX_INTERVAL_MS = 10_000; // 10 seconds
const FILE_QUEUE_INTERVAL_MS = 30_000; // 30 seconds
const HEALTH_CHECK_INTERVAL_MS = 3_600_000; // 1 hour
const JANITOR_INTERVAL_MS = 3_600_000; // 1 hour
const OUTBOX_BATCH_LIMIT = 50;
const FILE_QUEUE_BATCH_LIMIT = 5;
@@ -47,6 +49,7 @@ const PEER_UNREACHABLE_THRESHOLD = 10;
let outboxTimer: ReturnType<typeof setTimeout> | null = null;
let fileQueueTimer: ReturnType<typeof setTimeout> | null = null;
let healthCheckTimer: ReturnType<typeof setTimeout> | null = null;
let janitorTimer: ReturnType<typeof setTimeout> | null = null;
let outboxAbortController: AbortController | null = null;
let fileQueueAbortController: AbortController | null = null;
@@ -595,6 +598,15 @@ async function processHealthCheckTick(): Promise<void> {
}
}
// ─── Janitor Worker ──────────────────────────────────────────────────────────
function scheduleJanitorTick(): void {
janitorTimer = setTimeout(() => {
runFederationJanitor();
scheduleJanitorTick();
}, JANITOR_INTERVAL_MS);
}
// ─── Lifecycle ──────────────────────────────────────────────────────────────
/**
@@ -690,6 +702,7 @@ export function startFederationWorkers(): void {
scheduleOutboxTick();
scheduleFileQueueTick();
scheduleHealthCheckTick();
scheduleJanitorTick();
// Run initial sync for newly peered instances (async, non-blocking)
runInitialSyncForNewPeers().catch((err) => {
console.error('[federation-worker] Initial sync error:', err);
@@ -710,6 +723,11 @@ export function stopFederationWorkers(): void {
healthCheckTimer = null;
}
if (janitorTimer) {
clearTimeout(janitorTimer);
janitorTimer = null;
}
outboxAbortController?.abort();
outboxAbortController = null;
+141 -2
View File
@@ -1,9 +1,9 @@
import fs from 'fs';
import path from 'path';
import { and, eq, isNotNull, lt } from 'drizzle-orm';
import { and, eq, inArray, isNotNull, isNull, lt } from 'drizzle-orm';
import { config } from '../config.js';
import { getDb, getRawDb, schema } from '../db/index.js';
import { deleteUploadFile } from './fileCleanup.js';
import { deleteUploadFile, deleteAttachmentFiles } from './fileCleanup.js';
import type { StorageStats, StorageBreakdown, OrphanedFile, CleanupResult } from '@backspace/shared';
const IMAGE_EXTS = new Set(['.jpg', '.jpeg', '.png', '.gif', '.webp', '.svg', '.ico', '.bmp', '.avif']);
@@ -430,3 +430,142 @@ export function cleanupFederationFileQueue(): number {
return completed.changes + expired.changes;
}
/**
* Hard-delete DM channels that were soft-deleted more than 24 hours ago.
* Cascades: reactions, embeds, attachments (files + DB rows), messages,
* members, outbox entries, mutation log entries, file queue entries, then the channel.
* Returns the number of channels purged.
*/
export function cleanupSoftDeletedDmChannels(): number {
const db = getDb();
const gracePeriodMs = 24 * 60 * 60 * 1000; // 24 hours
const cutoff = Date.now() - gracePeriodMs;
const expired = db
.select({ id: schema.dmChannels.id })
.from(schema.dmChannels)
.where(
and(
isNotNull(schema.dmChannels.deletedAt),
lt(schema.dmChannels.deletedAt, cutoff),
),
)
.all();
if (expired.length === 0) return 0;
let purged = 0;
for (const channel of expired) {
try {
// Get message IDs and attachment filenames before deletion
const msgIds = db.select({ id: schema.dmMessages.id })
.from(schema.dmMessages)
.where(eq(schema.dmMessages.dmChannelId, channel.id))
.all()
.map(m => m.id);
const filesToDelete: string[] = [];
if (msgIds.length > 0) {
const attachments = db.select({ filename: schema.attachments.filename })
.from(schema.attachments)
.where(inArray(schema.attachments.dmMessageId, msgIds))
.all();
filesToDelete.push(...attachments.map(a => a.filename));
}
db.transaction((tx) => {
if (msgIds.length > 0) {
// Delete reactions
tx.delete(schema.dmReactions)
.where(inArray(schema.dmReactions.dmMessageId, msgIds))
.run();
// Delete embeds
tx.delete(schema.embeds)
.where(inArray(schema.embeds.dmMessageId, msgIds))
.run();
// Delete attachments (DB rows)
tx.delete(schema.attachments)
.where(inArray(schema.attachments.dmMessageId, msgIds))
.run();
// Delete file queue entries
tx.delete(schema.federationFileQueue)
.where(inArray(schema.federationFileQueue.dmMessageId, msgIds))
.run();
}
// Delete messages
tx.delete(schema.dmMessages)
.where(eq(schema.dmMessages.dmChannelId, channel.id))
.run();
// Delete members (should be 0, defensive)
tx.delete(schema.dmMembers)
.where(eq(schema.dmMembers.dmChannelId, channel.id))
.run();
// Delete read states
tx.delete(schema.readStates)
.where(eq(schema.readStates.channelId, channel.id))
.run();
// Delete federation outbox entries
tx.delete(schema.federationOutbox)
.where(eq(schema.federationOutbox.dmChannelId, channel.id))
.run();
// Delete mutation log entries
tx.delete(schema.federationMutationLog)
.where(eq(schema.federationMutationLog.dmChannelId, channel.id))
.run();
// Delete the channel itself
tx.delete(schema.dmChannels)
.where(eq(schema.dmChannels.id, channel.id))
.run();
});
// Clean up files from disk (outside transaction — filesystem ops are idempotent)
deleteAttachmentFiles(filesToDelete.map(f => ({ filename: f })));
purged++;
} catch (err) {
console.error(`[storage-janitor] Failed to purge soft-deleted DM channel ${channel.id}:`, err);
}
}
if (purged > 0) {
console.log(`[storage-janitor] Purged ${purged} soft-deleted DM channels`);
}
return purged;
}
/**
* Run all periodic federation/GC cleanup tasks:
* - Expired outbox entries
* - Old mutation log entries
* - Stale file queue entries
* - Soft-deleted DM channels past grace period
*/
export function runFederationJanitor(): void {
try {
const outbox = cleanupFederationOutbox();
const mutLog = cleanupFederationMutationLog();
const fileQ = cleanupFederationFileQueue();
const dmGc = cleanupSoftDeletedDmChannels();
const total = outbox + mutLog + fileQ + dmGc;
if (total > 0) {
console.log(
`[storage-janitor] Federation GC sweep: outbox=${outbox} mutationLog=${mutLog} fileQueue=${fileQ} dmChannels=${dmGc}`,
);
}
} catch (err) {
console.error('[storage-janitor] Federation GC sweep error:', err);
}
}
+3 -3
View File
@@ -2,7 +2,7 @@ import type { FastifyInstance } from 'fastify';
import type { WebSocket } from 'ws';
import { verifyJwt } from '../utils/auth.js';
import { getDb, schema } from '../db/index.js';
import { eq, and, inArray, desc, sql } from 'drizzle-orm';
import { eq, and, inArray, isNull, desc, sql } from 'drizzle-orm';
import { handleClientEvent } from './events.js';
import { computePermissions, PermissionBits, permissionsToString } from '../utils/permissions.js';
import type {
@@ -1050,10 +1050,10 @@ function buildReadyPayload(userId: string): {
const dmChannels: DmChannel[] = [];
if (dmChannelIds.length > 0) {
// Batch: all DM channels (1 query)
// Batch: all DM channels (1 query, exclude soft-deleted)
const allDmChannelRows = batchInArray(
dmChannelIds,
ids => db.select().from(schema.dmChannels).where(inArray(schema.dmChannels.id, ids)).all(),
ids => db.select().from(schema.dmChannels).where(and(inArray(schema.dmChannels.id, ids), isNull(schema.dmChannels.deletedAt))).all(),
);
const dmChannelMap = new Map(allDmChannelRows.map(c => [c.id, c]));