diff --git a/packages/server/src/routes/dm.ts b/packages/server/src/routes/dm.ts index d116ec7e..cbc09667 100644 --- a/packages/server/src/routes/dm.ts +++ b/packages/server/src/routes/dm.ts @@ -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 { 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 { .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 { } // 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 { } // 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 { }); } } 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 diff --git a/packages/server/src/utils/federationWorker.ts b/packages/server/src/utils/federationWorker.ts index fbf05806..c15933fc 100644 --- a/packages/server/src/utils/federationWorker.ts +++ b/packages/server/src/utils/federationWorker.ts @@ -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 | null = null; let fileQueueTimer: ReturnType | null = null; let healthCheckTimer: ReturnType | null = null; +let janitorTimer: ReturnType | null = null; let outboxAbortController: AbortController | null = null; let fileQueueAbortController: AbortController | null = null; @@ -595,6 +598,15 @@ async function processHealthCheckTick(): Promise { } } +// ─── 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; diff --git a/packages/server/src/utils/storageJanitor.ts b/packages/server/src/utils/storageJanitor.ts index 5e67f6a6..50efa356 100644 --- a/packages/server/src/utils/storageJanitor.ts +++ b/packages/server/src/utils/storageJanitor.ts @@ -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); + } +} diff --git a/packages/server/src/ws/handler.ts b/packages/server/src/ws/handler.ts index 64bc2748..4d3d0aa0 100644 --- a/packages/server/src/ws/handler.ts +++ b/packages/server/src/ws/handler.ts @@ -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]));