diff --git a/.env.example b/.env.example index 04b4295e8..7414cb5f1 100644 --- a/.env.example +++ b/.env.example @@ -304,6 +304,14 @@ WEBHOOK_SSRF_PROTECT=true # Per-file cap on status (Story) media persisted to disk/S3. A status whose media exceeds this is # stored omitted (mediaOmitted=true) rather than rejected. A non-positive/garbage value falls back. # STATUS_MEDIA_MAX_BYTES=10485760 # default 10 MiB +# Orphaned status media reconciliation: how often the sweep re-lists status media files to find +# ones no row references (crash leftovers), and how long a file must be seen unreferenced before +# it is deleted. Non-positive/garbage values fall back to the defaults. +# STATUS_ORPHAN_SWEEP_INTERVAL_MS=3600000 # default 1h +# STATUS_ORPHAN_GRACE_MS=3600000 # default 1h +# Cap on the final rendered text of a send-template request (after variable substitution). Over-cap +# renders are rejected with 400, never silently truncated. A non-positive/garbage value falls back. +# TEMPLATE_RENDER_MAX_CHARS=65536 # default 64 KiB # Cap on concurrently-running bulk batches (POST .../messages/send-bulk). 0 = unlimited. A non-finite # or negative value falls back to the default. # BULK_MAX_CONCURRENT_BATCHES=50 # default 50 diff --git a/openapi.json b/openapi.json index 8c01c86ba..7c9fc9d33 100644 --- a/openapi.json +++ b/openapi.json @@ -7489,6 +7489,7 @@ }, "participants": { "description": "Participant WhatsApp IDs (e.g. 628123456789@c.us)", + "maxItems": 256, "type": "array", "items": { "type": "string" @@ -7505,6 +7506,7 @@ "properties": { "participants": { "description": "Participant WhatsApp IDs (e.g. 628123456789@c.us)", + "maxItems": 256, "type": "array", "items": { "type": "string" diff --git a/src/config/configuration.ts b/src/config/configuration.ts index fcc4e4cfc..190c894a1 100644 --- a/src/config/configuration.ts +++ b/src/config/configuration.ts @@ -260,6 +260,30 @@ export default () => ({ const n = parseInt(process.env.STATUS_MEDIA_MAX_BYTES ?? '', 10); return Number.isFinite(n) && n > 0 ? n : 10 * 1024 * 1024; })(), + // How often the reconciliation sweep re-lists status media files to find ones no row references + // (default 1h). Orphans only arise from a crash between the media write and its row update. + orphanSweepIntervalMs: (() => { + const n = parseInt(process.env.STATUS_ORPHAN_SWEEP_INTERVAL_MS ?? '', 10); + return Number.isFinite(n) && n > 0 ? n : 60 * 60 * 1000; + })(), + // How long an unreferenced status media file must be observed by the sweep before it is deleted + // (default 1h), so a file mid-ingest is never reaped. A non-positive/garbage value falls back. + orphanGraceMs: (() => { + const n = parseInt(process.env.STATUS_ORPHAN_GRACE_MS ?? '', 10); + return Number.isFinite(n) && n > 0 ? n : 60 * 60 * 1000; + })(), + }, + + // Message-template rendering + template: { + // Cap on the FINAL rendered text of a send-template request (header+body+footer joined, after + // caller-supplied variable substitution; default 64 KiB — also WhatsApp's own text-message + // ceiling). Without it a small template plus a huge variable inflates the payload to the + // engine/DB unboundedly. Over-cap renders are rejected (400), never silently truncated. + renderMaxChars: (() => { + const n = parseInt(process.env.TEMPLATE_RENDER_MAX_CHARS ?? '', 10); + return Number.isFinite(n) && n > 0 ? n : 64 * 1024; + })(), }, // Storage configuration diff --git a/src/modules/group/dto/group.dto.spec.ts b/src/modules/group/dto/group.dto.spec.ts index db17d0bde..50036b36a 100644 --- a/src/modules/group/dto/group.dto.spec.ts +++ b/src/modules/group/dto/group.dto.spec.ts @@ -41,6 +41,20 @@ describe('group DTO validation', () => { expect(errors.length).toBeGreaterThan(0); }); + it('accepts a participants array at the batch cap (256)', async () => { + const participants = Array.from({ length: 256 }, (_, i) => `628100000${String(i).padStart(3, '0')}@c.us`); + expect(await errorsFor(ParticipantsDto, { participants })).toHaveLength(0); + expect(await errorsFor(CreateGroupDto, { name: 'G', participants })).toHaveLength(0); + }); + + it('rejects a participants array beyond the batch cap (the engine works the list sequentially)', async () => { + const participants = Array.from({ length: 257 }, (_, i) => `628100000${String(i).padStart(3, '0')}@c.us`); + const batchErrors = await errorsFor(ParticipantsDto, { participants }); + expect(batchErrors.some(e => e.property === 'participants')).toBe(true); + const createErrors = await errorsFor(CreateGroupDto, { name: 'G', participants }); + expect(createErrors.some(e => e.property === 'participants')).toBe(true); + }); + it('still rejects unknown properties (forbidNonWhitelisted intact)', async () => { const errors = await errorsFor(ParticipantsDto, { participants: ['x@c.us'], hacker: true }); expect(errors.some(e => e.property === 'hacker')).toBe(true); diff --git a/src/modules/group/dto/group.dto.ts b/src/modules/group/dto/group.dto.ts index e7730efd0..de5e885db 100644 --- a/src/modules/group/dto/group.dto.ts +++ b/src/modules/group/dto/group.dto.ts @@ -2,6 +2,7 @@ import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger'; import { IsArray, ArrayNotEmpty, + ArrayMaxSize, IsString, IsNotEmpty, MaxLength, @@ -17,6 +18,15 @@ import { ToStrictBoolean, ToStrictNumber } from '../../../common/utils/strict-bo export const GROUP_NAME_MAX_LENGTH = 100; export const GROUP_DESCRIPTION_MAX_LENGTH = 1024; +/** + * Max participants accepted in one group-create or participant-batch request. The engine works the + * list sequentially (one WhatsApp round-trip per participant, with an honest per-participant + * result), so an unbounded array turns a single HTTP call into minutes of serial engine work. + * 256 stays well inside WhatsApp's own group-size limit while keeping one request's work bounded; + * larger imports batch across requests. + */ +export const GROUP_PARTICIPANTS_MAX = 256; + export class CreateGroupDto { @ApiProperty({ description: 'Group subject/name', maxLength: GROUP_NAME_MAX_LENGTH }) @IsString() @@ -24,17 +34,27 @@ export class CreateGroupDto { @MaxLength(GROUP_NAME_MAX_LENGTH) name: string; - @ApiProperty({ description: 'Participant WhatsApp IDs (e.g. 628123456789@c.us)', type: [String] }) + @ApiProperty({ + description: 'Participant WhatsApp IDs (e.g. 628123456789@c.us)', + type: [String], + maxItems: GROUP_PARTICIPANTS_MAX, + }) @IsArray() @ArrayNotEmpty() + @ArrayMaxSize(GROUP_PARTICIPANTS_MAX) @IsString({ each: true }) participants: string[]; } export class ParticipantsDto { - @ApiProperty({ description: 'Participant WhatsApp IDs (e.g. 628123456789@c.us)', type: [String] }) + @ApiProperty({ + description: 'Participant WhatsApp IDs (e.g. 628123456789@c.us)', + type: [String], + maxItems: GROUP_PARTICIPANTS_MAX, + }) @IsArray() @ArrayNotEmpty() + @ArrayMaxSize(GROUP_PARTICIPANTS_MAX) @IsString({ each: true }) participants: string[]; } diff --git a/src/modules/group/group.service.spec.ts b/src/modules/group/group.service.spec.ts index f59f1996d..6c02546d3 100644 --- a/src/modules/group/group.service.spec.ts +++ b/src/modules/group/group.service.spec.ts @@ -1,8 +1,9 @@ -import { BadRequestException, NotFoundException } from '@nestjs/common'; +import { BadRequestException, HttpException, NotFoundException } from '@nestjs/common'; import { GroupService } from './group.service'; import { SessionService } from '../session/session.service'; import { IWhatsAppEngine } from '../../engine/interfaces/whatsapp-engine.interface'; import { EngineNotSupportedError } from '../../common/errors/engine-not-supported.error'; +import { EngineRefusedError } from '../../common/errors/engine-refused.error'; describe('GroupService', () => { const makeService = (engine: Partial | undefined) => { @@ -149,5 +150,37 @@ describe('GroupService', () => { expect(engine.setGroupMessagesAdminsOnly).not.toHaveBeenCalled(); expect(engine.setGroupInfoAdminsOnly).not.toHaveBeenCalled(); }); + + it('names the failed field AND the applied ones when a patch partially applies', async () => { + // ephemeralSeconds applied, then announce failed: the client must learn the group is now in a + // mixed state (and which subset took effect), not receive a bare engine error. + const engine = { + setGroupEphemeral: jest.fn().mockResolvedValue(undefined), + setGroupMessagesAdminsOnly: jest.fn().mockRejectedValue(new EngineRefusedError('not a group admin')), + setGroupInfoAdminsOnly: jest.fn().mockResolvedValue(undefined), + }; + const svc = makeService(engine); + const error = await svc + .updateGroupSettings('s1', 'g1', { announce: true, locked: true, ephemeralSeconds: 86400 }) + .catch((e: unknown) => e); + expect(error).toBeInstanceOf(HttpException); + expect((error as HttpException).getStatus()).toBe(403); // the underlying refusal's status survives + expect((error as HttpException).message).toContain("'announce' failed"); + expect((error as HttpException).message).toContain('ephemeralSeconds'); + // The patch stops at the failure: locked is never attempted. + expect(engine.setGroupInfoAdminsOnly).not.toHaveBeenCalled(); + }); + + it('propagates a first-field failure unchanged (nothing applied → no partial state to report)', async () => { + const engine = { + setGroupEphemeral: jest.fn().mockRejectedValue(new EngineRefusedError('not a group admin')), + setGroupMessagesAdminsOnly: jest.fn().mockResolvedValue(undefined), + }; + const svc = makeService(engine); + await expect( + svc.updateGroupSettings('s1', 'g1', { announce: true, ephemeralSeconds: 86400 }), + ).rejects.toBeInstanceOf(EngineRefusedError); + expect(engine.setGroupMessagesAdminsOnly).not.toHaveBeenCalled(); + }); }); }); diff --git a/src/modules/group/group.service.ts b/src/modules/group/group.service.ts index 3f3a3679a..27d1ab31d 100644 --- a/src/modules/group/group.service.ts +++ b/src/modules/group/group.service.ts @@ -1,4 +1,4 @@ -import { Injectable, BadRequestException, NotFoundException } from '@nestjs/common'; +import { Injectable, BadRequestException, HttpException, HttpStatus, NotFoundException } from '@nestjs/common'; import { SessionService } from '../session/session.service'; import { IWhatsAppEngine } from '../../engine/interfaces/whatsapp-engine.interface'; import { paginate, ListOptions } from '../../common/utils/paginate'; @@ -98,6 +98,11 @@ export class GroupService { * Ordering matters: ephemeralSeconds is applied FIRST because it is the only field with a * deterministic per-engine refusal (wwjs always 501s it). Applying announce/locked first would * leave a silently half-applied patch behind when the ephemeral call then throws. + * + * A failure on the FIRST applied field propagates unchanged (nothing was applied, so the patch + * simply failed). A failure on a LATER field means the group is now in a mixed state, so the + * error names the failed field and the ones already applied — the caller can reconcile instead + * of guessing which subset took effect. The wrapped error keeps the underlying HTTP status. */ async updateGroupSettings( sessionId: string, @@ -109,14 +114,33 @@ export class GroupService { throw new BadRequestException('At least one of announce, locked, ephemeralSeconds must be provided'); } const engine = this.getEngine(sessionId); + const steps: Array<[field: string, apply: () => Promise]> = []; if (ephemeralSeconds !== undefined) { - await engine.setGroupEphemeral(groupId, ephemeralSeconds); + steps.push(['ephemeralSeconds', () => engine.setGroupEphemeral(groupId, ephemeralSeconds)]); } if (announce !== undefined) { - await engine.setGroupMessagesAdminsOnly(groupId, announce); + steps.push(['announce', () => engine.setGroupMessagesAdminsOnly(groupId, announce)]); } if (locked !== undefined) { - await engine.setGroupInfoAdminsOnly(groupId, locked); + steps.push(['locked', () => engine.setGroupInfoAdminsOnly(groupId, locked)]); + } + + const applied: string[] = []; + for (const [field, apply] of steps) { + try { + await apply(); + applied.push(field); + } catch (error) { + if (applied.length === 0) throw error; + const status = error instanceof HttpException ? error.getStatus() : HttpStatus.INTERNAL_SERVER_ERROR; + const detail = error instanceof Error ? error.message : String(error); + throw new HttpException( + `Group settings only partially applied: '${field}' failed (${detail}); already applied: ${applied.join( + ', ', + )}`, + status, + ); + } } } } diff --git a/src/modules/message/message.service.spec.ts b/src/modules/message/message.service.spec.ts index 2e15549b7..1c4de0924 100644 --- a/src/modules/message/message.service.spec.ts +++ b/src/modules/message/message.service.spec.ts @@ -359,6 +359,52 @@ describe('MessageService', () => { ); expect(mockEngine.sendTextMessage).not.toHaveBeenCalled(); }); + + it('rejects an over-cap render with a 400 naming the limit (never truncated silently)', async () => { + (templateService.resolve as jest.Mock).mockResolvedValue(mockTemplate({ body: 'Hi {{customer}}' })); + + const error = await service + .sendTemplate('sess-1', { + chatId: 'test@c.us', + templateId: 'tpl-1', + vars: { customer: 'x'.repeat(70_000) }, + }) + .catch((e: unknown) => e); + + expect(error).toBeInstanceOf(BadRequestException); + expect((error as Error).message).toContain('over the 65536-character limit'); + expect(mockEngine.sendTextMessage).not.toHaveBeenCalled(); + }); + + it('renders at-or-under the cap unchanged', async () => { + (templateService.resolve as jest.Mock).mockResolvedValue(mockTemplate({ body: 'Hi {{customer}}' })); + // 'Hi ' + name lands exactly on the 64 KiB default cap — at the cap is NOT over it. + const name = 'y'.repeat(64 * 1024 - 3); + + await service.sendTemplate('sess-1', { chatId: 'test@c.us', templateId: 'tpl-1', vars: { customer: name } }); + + expect(mockEngine.sendTextMessage).toHaveBeenCalledWith('test@c.us', `Hi ${name}`); + }); + + it('honors a configured template.renderMaxChars override', async () => { + const configService = { + get: (key: string, fallback: unknown) => (key === 'template.renderMaxChars' ? 10 : fallback), + } as unknown as ConstructorParameters[5]; + const capped = new MessageService( + repository as Repository, + sessionService as unknown as SessionService, + hookManager as HookManager, + templateService as unknown as TemplateService, + lidMappingStore as unknown as LidMappingStoreService, + configService, + ); + (templateService.resolve as jest.Mock).mockResolvedValue(mockTemplate({ body: 'Hello {{customer}}' })); + + await expect( + capped.sendTemplate('sess-1', { chatId: 'test@c.us', templateId: 'tpl-1', vars: { customer: 'Alice' } }), + ).rejects.toThrow(/over the 10-character limit/); + expect(mockEngine.sendTextMessage).not.toHaveBeenCalled(); + }); }); // ── send-hook chokepoint ────────────────────────────────────────── diff --git a/src/modules/message/message.service.ts b/src/modules/message/message.service.ts index 75b346197..26047f22f 100644 --- a/src/modules/message/message.service.ts +++ b/src/modules/message/message.service.ts @@ -27,6 +27,9 @@ export interface GetMessagesOptions { offset?: number; } +/** Default cap on a rendered template's final text; overridable via TEMPLATE_RENDER_MAX_CHARS. */ +export const DEFAULT_TEMPLATE_RENDER_MAX_CHARS = 64 * 1024; + /** * Outbound sends are executed directly against the WhatsApp engine, not via a BullMQ queue. * @@ -141,6 +144,10 @@ export class MessageService { * existing {@link sendText} path so plugin hooks, persistence, and status * tracking are reused. Throws NotFoundException when the template cannot be * resolved by id or name. + * + * The FINAL rendered text is capped at template.renderMaxChars (default 64 KiB): caller-supplied + * variables can inflate a small template unboundedly, so an over-cap render is rejected with a + * 400 naming the limit rather than truncated silently or pushed to the engine/DB as-is. */ async sendTemplate(sessionId: string, dto: SendTemplateMessageDto): Promise { const template = await this.templateService.resolve(sessionId, { @@ -154,6 +161,15 @@ export class MessageService { .map(segment => renderTemplate(segment, vars)); const text = segments.join('\n\n'); + const maxChars = + this.configService?.get('template.renderMaxChars', DEFAULT_TEMPLATE_RENDER_MAX_CHARS) ?? + DEFAULT_TEMPLATE_RENDER_MAX_CHARS; + if (text.length > maxChars) { + throw new BadRequestException( + `Rendered template is ${text.length} characters, over the ${maxChars}-character limit`, + ); + } + return this.sendText(sessionId, { chatId: dto.chatId, text }); } diff --git a/src/modules/status-store/status-store.service.spec.ts b/src/modules/status-store/status-store.service.spec.ts index 8d85b3820..d4c22cbd7 100644 --- a/src/modules/status-store/status-store.service.spec.ts +++ b/src/modules/status-store/status-store.service.spec.ts @@ -258,10 +258,14 @@ describe('StatusStoreService ingest race (unique-constraint loser)', () => { fs.rmSync(baseDir, { recursive: true, force: true }); }); - const mediaDir = (): string[] => fs.readdirSync(path.join(baseDir, 'media', 'statuses', 'sess')); + const mediaDir = (): string[] => { + const dir = path.join(baseDir, 'media', 'statuses', 'sess'); + return fs.existsSync(dir) ? fs.readdirSync(dir) : []; + }; - it('reaps the media file it wrote when a concurrent ingest wins the unique constraint', async () => { - // The winner committed its own media file; the loser must not leave a second, orphaned one. + it('never writes a media file when a concurrent ingest wins the unique constraint (row-first)', async () => { + // The winner committed its own media file; with row-first ordering the loser has not written + // anything at the point its save fails, so there is nothing to reap and no second file. fs.mkdirSync(path.join(baseDir, 'media', 'statuses', 'sess'), { recursive: true }); fs.writeFileSync(path.join(baseDir, 'media', 'statuses', 'sess', 'winner.jpg'), 'winner'); const winner = new StatusUpdate(); @@ -286,7 +290,7 @@ describe('StatusStoreService ingest race (unique-constraint loser)', () => { expect(mediaDir()).toEqual(['winner.jpg']); }); - it('reaps the file it wrote when the save fails with no winner row, then rethrows', async () => { + it('writes no file when the row save fails with no winner row, then rethrows', async () => { const repo = { findOne: jest.fn().mockResolvedValue(null), save: jest.fn().mockRejectedValue(new Error('database is locked')), @@ -303,26 +307,21 @@ describe('StatusStoreService ingest race (unique-constraint loser)', () => { }), ).rejects.toThrow('database is locked'); + // Row-first: the save precedes any file write, so a failed save can never leak an orphan. expect(mediaDir()).toHaveLength(0); }); - it('keeps the file but still rethrows when the re-read row is this very insert (driver errored on a commit that landed)', async () => { + it('rethrows a non-unique save error even when the commit landed, leaving no orphan file behind', async () => { + // The pathological "driver errored on a commit that landed" case: the re-read finds a row, but + // the failure is genuine (not a unique violation) and must surface. The landed row carries no + // media reference (the file write comes after the save), so the state stays consistent. + const landed = new StatusUpdate(); const repo = { - findOne: jest - .fn() - .mockResolvedValueOnce(null) - // The row that "landed" references exactly the file this ingest just wrote. - .mockImplementation(() => { - const selfRow = new StatusUpdate(); - selfRow.mediaPath = `statuses/sess/${mediaDir()[0]}`; - return Promise.resolve(selfRow); - }), + findOne: jest.fn().mockResolvedValueOnce(null).mockResolvedValue(landed), save: jest.fn().mockRejectedValue(new Error('driver reported failure after commit')), } as unknown as Repository; const service = new StatusStoreService(repo, storageService, fakeConfigService()); - // Not a unique-constraint error, so the failure surfaces even though a row exists — but the - // file is kept: the re-read row IS this insert, so its media is still referenced. await expect( service.ingest('sess', { waStatusId: 'raced', @@ -333,7 +332,7 @@ describe('StatusStoreService ingest race (unique-constraint loser)', () => { }), ).rejects.toThrow('driver reported failure after commit'); - expect(mediaDir()).toHaveLength(1); + expect(mediaDir()).toHaveLength(0); }); it('rethrows a non-unique save error even when a coincidental winner row exists', async () => { @@ -348,7 +347,7 @@ describe('StatusStoreService ingest race (unique-constraint loser)', () => { const service = new StatusStoreService(repo, storageService, fakeConfigService()); // A genuine persistence failure must not be swallowed into an idempotent return just because - // a matching row happens to exist — and this call's own file is still reaped. + // a matching row happens to exist — and this call never wrote a file to begin with. await expect( service.ingest('sess', { waStatusId: 'raced', @@ -506,13 +505,143 @@ describe('StatusStoreService.purgeExpired', () => { expect(removed).toBe(0); expect(await repository.count()).toBe(1); }); + + it('keeps a row whose media delete failed (retried next sweep), still purging the rest', async () => { + const expiredWithMedia = await ingestWithMedia('expired-media', 1000); + await service.ingest('sess', { + waStatusId: 'expired-text', + contactJid: '628111@c.us', + type: 'text', + postedAt: 2000, + }); + const mediaFile = path.join(baseDir, 'media', expiredWithMedia.mediaPath!); + + // A backend outage fails the file delete: the row must survive (deleting it would orphan the + // file permanently), while the file-less text row is still purged. + const failingStorage = { + deleteFile: jest.fn().mockRejectedValue(new Error('backend down')), + } as unknown as StorageService; + const failingService = new StatusStoreService(repository, failingStorage, fakeConfigService()); + + const now = 2000 + 24 * 60 * 60 * 1000 + 1; + const removed = await failingService.purgeExpired(now); + + expect(removed).toBe(1); // only the text row (no file to delete) + const remaining = await repository.find(); + expect(remaining.map(r => r.waStatusId)).toEqual(['expired-media']); + expect(fs.existsSync(mediaFile)).toBe(true); + + // The next sweep, with a healthy backend, finishes the job. + const retried = await service.purgeExpired(now); + expect(retried).toBe(1); + expect(await repository.count()).toBe(0); + expect(fs.existsSync(mediaFile)).toBe(false); + }); +}); + +describe('StatusStoreService.sweepOrphanedMedia', () => { + let baseDir: string; + let ds: DataSource; + let repository: Repository; + let storageService: StorageService; + let service: StatusStoreService; + + beforeEach(async () => { + baseDir = fs.mkdtempSync(path.join(os.tmpdir(), 'owa-status-sweep-')); + ds = new DataSource({ type: 'better-sqlite3', database: ':memory:', entities: [StatusUpdate], synchronize: true }); + await ds.initialize(); + repository = ds.getRepository(StatusUpdate); + storageService = makeStorageService(path.join(baseDir, 'media')); + service = new StatusStoreService(repository, storageService, fakeConfigService()); + }); + + afterEach(async () => { + if (ds.isInitialized) await ds.destroy(); + fs.rmSync(baseDir, { recursive: true, force: true }); + }); + + const ingestWithMedia = async (waStatusId: string, postedAt: number): Promise => + ( + await service.ingest('sess', { + waStatusId, + contactJid: '628111@c.us', + type: 'image', + media: { mimetype: 'image/jpeg', data: Buffer.from(waStatusId).toString('base64') }, + postedAt, + }) + ).row; + + const writeFile = (key: string, contents: string): void => { + const full = path.join(baseDir, 'media', key); + fs.mkdirSync(path.dirname(full), { recursive: true }); + fs.writeFileSync(full, contents); + }; + + it('reaps an orphan only after the grace window; never touches referenced or non-status files', async () => { + const live = await ingestWithMedia('live', Date.now()); + const liveFile = path.join(baseDir, 'media', live.mediaPath!); + // A file with no referencing row — the crash-between-write-and-row-update leftover. + writeFile('statuses/sess/orphan.jpg', 'orphan'); + // Chat media shares the store; the sweep is scoped to the statuses/ prefix. + writeFile('chat/sess/keep.jpg', 'keep'); + + const t0 = Date.now(); + // First sighting only records the orphan; inside the grace window nothing is deleted. + expect(await service.sweepOrphanedMedia(t0)).toBe(0); + expect(await service.sweepOrphanedMedia(t0 + 30 * 60 * 1000)).toBe(0); + expect(fs.existsSync(path.join(baseDir, 'media', 'statuses', 'sess', 'orphan.jpg'))).toBe(true); + + // Past the grace window (default 1h) the orphan is reaped; referenced and non-status files stay. + expect(await service.sweepOrphanedMedia(t0 + 61 * 60 * 1000)).toBe(1); + expect(fs.existsSync(path.join(baseDir, 'media', 'statuses', 'sess', 'orphan.jpg'))).toBe(false); + expect(fs.existsSync(liveFile)).toBe(true); + expect(fs.existsSync(path.join(baseDir, 'media', 'chat', 'sess', 'keep.jpg'))).toBe(true); + }); + + it('never reaps a file that becomes referenced between sightings', async () => { + // First pass sees the file while its row update has not landed yet (the ingest crash window). + writeFile('statuses/sess/pending.jpg', 'pending'); + const t0 = Date.now(); + expect(await service.sweepOrphanedMedia(t0)).toBe(0); + + // The row update lands, now referencing the file. + const row = new StatusUpdate(); + row.sessionId = 'sess'; + row.contactJid = '628111@c.us'; + row.waStatusId = 'late-reference'; + row.type = 'image'; + row.postedAt = t0; + row.expiresAt = t0 + 24 * 60 * 60 * 1000; + row.mediaPath = 'statuses/sess/pending.jpg'; + row.mediaMimetype = 'image/jpeg'; + row.mediaOmitted = false; + await repository.save(row); + + // Even past the grace window the file is safe — the referenced check re-reads the rows. + expect(await service.sweepOrphanedMedia(t0 + 61 * 60 * 1000)).toBe(0); + expect(fs.existsSync(path.join(baseDir, 'media', 'statuses', 'sess', 'pending.jpg'))).toBe(true); + }); + + it('honors a configured status.orphanGraceMs override', async () => { + const shortGrace = new StatusStoreService( + repository, + storageService, + fakeConfigService({ 'status.orphanGraceMs': 1000 }), + ); + writeFile('statuses/sess/orphan.jpg', 'orphan'); + + const t0 = Date.now(); + expect(await shortGrace.sweepOrphanedMedia(t0)).toBe(0); + expect(await shortGrace.sweepOrphanedMedia(t0 + 1001)).toBe(1); + expect(fs.existsSync(path.join(baseDir, 'media', 'statuses', 'sess', 'orphan.jpg'))).toBe(false); + }); }); -describe('StatusStoreService onModuleInit/onModuleDestroy (purge scheduling)', () => { +describe('StatusStoreService onModuleInit/onModuleDestroy (sweep scheduling)', () => { const mockDeps = (): { repo: Repository; storage: StorageService; find: jest.Mock } => { const find = jest.fn().mockResolvedValue([]); const repo = { find } as unknown as Repository; - const storage = {} as StorageService; + const storage = { listFiles: jest.fn().mockResolvedValue([]) } as unknown as StorageService; return { repo, storage, find }; }; @@ -538,4 +667,51 @@ describe('StatusStoreService onModuleInit/onModuleDestroy (purge scheduling)', ( jest.useRealTimers(); } }); + + it('runs the orphan sweep once at startup and on its own (slower) cadence, cleared on destroy', () => { + const { repo, storage } = mockDeps(); + const service = new StatusStoreService(repo, storage, fakeConfigService()); + + jest.useFakeTimers(); + try { + const sweepSpy = jest.spyOn(service, 'sweepOrphanedMedia').mockResolvedValue(0); + service.onModuleInit(); + expect(sweepSpy).toHaveBeenCalledTimes(1); + + // The TTL purge fires every 15 min; the orphan sweep only on its own 1h interval. + sweepSpy.mockClear(); + jest.advanceTimersByTime(15 * 60 * 1000); + expect(sweepSpy).not.toHaveBeenCalled(); + jest.advanceTimersByTime(45 * 60 * 1000); + expect(sweepSpy).toHaveBeenCalledTimes(1); + + service.onModuleDestroy(); + sweepSpy.mockClear(); + jest.advanceTimersByTime(60 * 60 * 1000); + expect(sweepSpy).not.toHaveBeenCalled(); + } finally { + jest.useRealTimers(); + } + }); + + it('honors a configured status.orphanSweepIntervalMs override', () => { + const { repo, storage } = mockDeps(); + const service = new StatusStoreService( + repo, + storage, + fakeConfigService({ 'status.orphanSweepIntervalMs': 5 * 60 * 1000 }), + ); + + jest.useFakeTimers(); + try { + const sweepSpy = jest.spyOn(service, 'sweepOrphanedMedia').mockResolvedValue(0); + service.onModuleInit(); + sweepSpy.mockClear(); + jest.advanceTimersByTime(5 * 60 * 1000); + expect(sweepSpy).toHaveBeenCalledTimes(1); + service.onModuleDestroy(); + } finally { + jest.useRealTimers(); + } + }); }); diff --git a/src/modules/status-store/status-store.service.ts b/src/modules/status-store/status-store.service.ts index 2f5adf1eb..97963687d 100644 --- a/src/modules/status-store/status-store.service.ts +++ b/src/modules/status-store/status-store.service.ts @@ -1,7 +1,7 @@ import { Injectable, OnModuleDestroy, OnModuleInit, Optional } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; import { ConfigService } from '@nestjs/config'; -import { In, LessThan, MoreThan, Repository } from 'typeorm'; +import { In, IsNull, LessThan, MoreThan, Not, Repository } from 'typeorm'; import { randomUUID } from 'crypto'; import { StatusUpdate } from './entities/status-update.entity'; import type { IncomingStatus } from './incoming-status'; @@ -20,6 +20,12 @@ const PURGE_INTERVAL_MS = 15 * 60 * 1000; /** Default per-file cap on persisted status media. Exported for the session service's seed, which * pre-gates history downloads at the same cap so over-cap blobs are never fetched. */ export const DEFAULT_MEDIA_MAX_BYTES = 10 * 1024 * 1024; +/** Default cadence of the orphaned-media reconciliation sweep (overridable via + * STATUS_ORPHAN_SWEEP_INTERVAL_MS). Lighter-than-it-sounds: one listFiles + one indexed query. */ +const DEFAULT_ORPHAN_SWEEP_INTERVAL_MS = 60 * 60 * 1000; +/** Default grace window before an unreferenced status media file is deleted (overridable via + * STATUS_ORPHAN_GRACE_MS), so a file mid-ingest is never reaped. */ +const DEFAULT_ORPHAN_GRACE_MS = 60 * 60 * 1000; /** Subtypes whose registered mimetype name differs from the conventional file extension. */ const MIME_SUBTYPE_EXT_OVERRIDES: Record = { jpeg: 'jpg', quicktime: 'mov' }; @@ -33,13 +39,17 @@ function extFromMimetype(mimetype: string): string { /** * Persists inbound status/story broadcasts (`StatusUpdate` rows) with a 24h TTL, storing any attached - * media through `StorageService`. Runs its own purge sweep (once at startup, then every 15 minutes) so - * the store never accumulates stories past their WhatsApp-side expiry. + * media through `StorageService`. Runs two recurring sweeps: the TTL purge (once at startup, then + * every 15 minutes) so the store never accumulates stories past their WhatsApp-side expiry, and an + * hourly reconciliation sweep that reaps media files no row references (crash leftovers). */ @Injectable() export class StatusStoreService implements OnModuleInit, OnModuleDestroy { private readonly logger = createLogger('StatusStoreService'); private purgeTimer?: ReturnType; + private orphanSweepTimer?: ReturnType; + /** First sweep sighting (epoch ms) of each unreferenced status media file, for the grace window. */ + private readonly orphanFirstSeenAt = new Map(); constructor( @InjectRepository(StatusUpdate, 'data') @@ -60,10 +70,24 @@ export class StatusStoreService implements OnModuleInit, OnModuleDestroy { runPurge(); // sweep once at startup this.purgeTimer = setInterval(runPurge, PURGE_INTERVAL_MS); this.purgeTimer.unref?.(); + + const runOrphanSweep = (): void => { + this.sweepOrphanedMedia(Date.now()).catch(err => + this.logger.error('Status media orphan sweep failed', err instanceof Error ? err.stack : String(err)), + ); + }; + runOrphanSweep(); // the first pass only records first-seen; nothing is deleted before the grace window + const sweepIntervalMs = this.configService.get( + 'status.orphanSweepIntervalMs', + DEFAULT_ORPHAN_SWEEP_INTERVAL_MS, + ); + this.orphanSweepTimer = setInterval(runOrphanSweep, sweepIntervalMs); + this.orphanSweepTimer.unref?.(); } onModuleDestroy(): void { if (this.purgeTimer) clearInterval(this.purgeTimer); + if (this.orphanSweepTimer) clearInterval(this.orphanSweepTimer); } /** @@ -71,6 +95,11 @@ export class StatusStoreService implements OnModuleInit, OnModuleDestroy { * `created` tells the caller whether this call actually inserted the row — false for a duplicate * delivery or a lost insert race — so a once-per-status side effect (the status.received webhook) * doesn't fire again for a status consumers already saw. + * + * Row-first ordering: the row is saved BEFORE its media file is written. A crash mid-ingest then + * leaves a consistent media-omitted row behind (the worst case is a missing attachment), never a + * permanent orphan file; the narrow window between the file write and the row update is the only + * remaining orphan source, and the reconciliation sweep reaps those. */ async ingest(sessionId: string, s: IncomingStatus): Promise<{ row: StatusUpdate; created: boolean }> { const existing = await this.repository.findOne({ where: { sessionId, waStatusId: s.waStatusId } }); @@ -88,37 +117,43 @@ export class StatusStoreService implements OnModuleInit, OnModuleDestroy { row.font = s.font; row.postedAt = s.postedAt; row.expiresAt = s.postedAt + STATUS_TTL_MS; - await this.attachMedia(row, sessionId, s); + // Settles every media outcome that does NOT need a file write; a keepable blob stays + // omitted-for-now until attachMedia lands the file and flips it (post-save). + const keepMedia = this.applyMediaDecision(row, s); + let saved: StatusUpdate; try { - return { row: await this.repository.save(row), created: true }; + saved = await this.repository.save(row); } catch (error) { // The unique (sessionId, waStatusId) index is the idempotency backstop for a concurrent ingest // of the same status racing past the findOne check above; on a unique-constraint violation the // loser re-reads and returns the row the winner just inserted instead of surfacing the // constraint error to the caller. Any OTHER save error is a genuine persistence failure and - // must propagate — never be masked by a coincidentally matching row. + // must propagate — never be masked by a coincidentally matching row. No media file has been + // written at this point (row-first ordering), so unlike a file-first ingest there is nothing + // to reap here — the winner's own call owns its file write. const winner = await this.repository.findOne({ where: { sessionId, waStatusId: s.waStatusId } }); - // This call's row was never saved, so the file attachMedia wrote for it (its own random key) - // is referenced by no row, and purgeExpired — which only sweeps expired rows' files — could - // never reap it. Delete it here or a failed ingest leaks one orphan per failure. The guard - // skips the pathological case where the re-read row is this very insert (driver errored on a - // commit that landed): there the file IS the persisted row's media. deleteFile treats an - // already-missing file as success. - if (row.mediaPath && row.mediaPath !== winner?.mediaPath) { - await this.storageService.deleteFile(row.mediaPath).catch(() => undefined); - } if (winner && isUniqueConstraintError(error)) return { row: winner, created: false }; throw error; } + + if (keepMedia) { + await this.attachMedia(saved, sessionId, s); + } + return { row: saved, created: true }; } - /** Sets the row's media* / mediaOmitted / omitReason fields, writing the file via StorageService when kept. */ - private async attachMedia(row: StatusUpdate, sessionId: string, s: IncomingStatus): Promise { + /** + * Settle the row's mediaOmitted/omitReason fields for every outcome that does not require a file + * write (no media, engine-omitted, over the store cap). Returns true when the blob should be + * written post-save; the row is then saved as omitted-for-now (no omitReason) and attachMedia + * upgrades it once the file exists. + */ + private applyMediaDecision(row: StatusUpdate, s: IncomingStatus): boolean { const media = s.media; if (!media) { row.mediaOmitted = false; - return; + return false; } const maxBytes = this.configService.get('status.mediaMaxBytes', DEFAULT_MEDIA_MAX_BYTES); @@ -126,22 +161,8 @@ export class StatusStoreService implements OnModuleInit, OnModuleDestroy { const withinCap = sizeBytes !== undefined && sizeBytes <= maxBytes; if (!media.omitted && media.data && withinCap) { - const key = `statuses/${sessionId}/${randomUUID()}.${extFromMimetype(media.mimetype)}`; - try { - await this.storageService.putFile(key, Buffer.from(media.data, 'base64')); - row.mediaPath = key; - row.mediaMimetype = media.mimetype; - row.mediaOmitted = false; - return; - } catch (error) { - this.logger.error( - `Failed to persist status media for session ${sessionId}, status ${s.waStatusId}`, - error instanceof Error ? error.stack : String(error), - ); - row.mediaOmitted = true; - row.omitReason = 'write_failed'; - return; - } + row.mediaOmitted = true; // pending the post-save write; attachMedia flips it back off + return true; } row.mediaOmitted = true; @@ -149,6 +170,52 @@ export class StatusStoreService implements OnModuleInit, OnModuleDestroy { // seed's pre-gate) is over the STORE's cap — keep the reason truthful on both arrival paths. row.omitReason = sizeBytes !== undefined && sizeBytes > maxBytes ? 'over_cap' : media.omitted ? 'engine_omitted' : 'over_cap'; + return false; + } + + /** + * Write the media file for a row already saved in the omitted-for-now state, then point the row at + * it. A failed write is recorded as omitReason 'write_failed' rather than thrown — a storage + * hiccup must not lose the status itself. A failure of the follow-up row update leaves the file + * unreferenced; that is the reconciliation sweep's job, so it is logged and swallowed here. + */ + private async attachMedia(row: StatusUpdate, sessionId: string, s: IncomingStatus): Promise { + const media = s.media; + if (!media?.data) return; + + const key = `statuses/${sessionId}/${randomUUID()}.${extFromMimetype(media.mimetype)}`; + try { + await this.storageService.putFile(key, Buffer.from(media.data, 'base64')); + } catch (error) { + this.logger.error( + `Failed to persist status media for session ${sessionId}, status ${s.waStatusId}`, + error instanceof Error ? error.stack : String(error), + ); + row.omitReason = 'write_failed'; + await this.repository.save(row).catch(err => + this.logger.warn(`Failed to record the write_failed omission for status ${s.waStatusId}`, { + error: String(err), + }), + ); + return; + } + + row.mediaPath = key; + row.mediaMimetype = media.mimetype; + row.mediaOmitted = false; + try { + await this.repository.save(row); + } catch (error) { + // The file exists but the row cannot reference it — an orphan the reconciliation sweep reaps + // after its grace window. The persisted row stays consistent (still omitted from the first + // save), so don't fail the ingest; restore the matching in-memory state too. + this.logger.warn(`Status media ${key} written but the row update failed; leaving it for the orphan sweep`, { + error: String(error), + }); + row.mediaPath = undefined; + row.mediaMimetype = undefined; + row.mediaOmitted = true; + } } // Reads exclude already-expired rows: WhatsApp hides a status the moment its 24h are up, but the @@ -220,24 +287,84 @@ export class StatusStoreService implements OnModuleInit, OnModuleDestroy { return { path: row.mediaPath, mimetype: row.mediaMimetype }; } - /** Deletes rows (and their media files) whose `expiresAt` is before `now`. Returns the count removed. */ + /** + * Deletes rows (and their media files) whose `expiresAt` is before `now`. Returns the count removed. + * A row whose media delete FAILS is kept — deleting it would orphan the file permanently (a row-less + * file is only rediscovered by the orphan sweep after its grace window), so the row stays and the + * next sweep retries the delete. A missing file is a successful delete (see StorageService), so an + * already-gone file never wedges its row. + */ async purgeExpired(now: number): Promise { const expired = await this.repository.find({ where: { expiresAt: LessThan(now) } }); if (expired.length === 0) return 0; + const deletableIds: string[] = []; await Promise.all( - expired - .filter((row): row is StatusUpdate & { mediaPath: string } => !!row.mediaPath) - .map(row => - this.storageService - .deleteFile(row.mediaPath) - .catch(err => - this.logger.warn(`Failed to delete expired status media ${row.mediaPath}`, { error: String(err) }), - ), - ), + expired.map(async row => { + if (!row.mediaPath) { + deletableIds.push(row.id); + return; + } + try { + await this.storageService.deleteFile(row.mediaPath); + deletableIds.push(row.id); + } catch (err) { + this.logger.warn( + `Failed to delete expired status media ${row.mediaPath}; keeping the row for the next sweep`, + { + error: String(err), + }, + ); + } + }), ); - const result = await this.repository.delete(expired.map(row => row.id)); - return result.affected ?? expired.length; + const result = await this.repository.delete(deletableIds); + return result.affected ?? deletableIds.length; + } + + /** + * Reconcile the media store with the status rows: delete `statuses/` files no row references. + * Orphans arise from the narrow crash window between the media write and its row update (row-first + * ingest keeps every other path consistent). Scoped to the statuses/ prefix — the media store is + * shared with chat media, which this sweep must never touch. + * + * A file is deleted only after the sweep has seen it unreferenced for at least the grace window: + * first-seen is tracked in memory (a restart simply restarts the grace clock, which fails safe), + * so a file mid-ingest or freshly written is never reaped. Returns the count removed. + */ + async sweepOrphanedMedia(now: number = Date.now()): Promise { + const graceMs = this.configService.get('status.orphanGraceMs', DEFAULT_ORPHAN_GRACE_MS); + const files = (await this.storageService.listFiles()).filter(file => file.startsWith('statuses/')); + const rows = await this.repository.find({ + where: { mediaPath: Not(IsNull()) }, + select: { mediaPath: true }, + }); + const referenced = new Set(rows.map(row => row.mediaPath)); + + let removed = 0; + const present = new Set(files); + for (const file of files) { + if (referenced.has(file)) { + this.orphanFirstSeenAt.delete(file); + continue; + } + const firstSeenAt = this.orphanFirstSeenAt.get(file) ?? now; + this.orphanFirstSeenAt.set(file, firstSeenAt); + if (now - firstSeenAt < graceMs) continue; + try { + await this.storageService.deleteFile(file); + this.orphanFirstSeenAt.delete(file); + removed += 1; + } catch (err) { + this.logger.warn(`Failed to delete orphaned status media ${file}`, { error: String(err) }); + } + } + // Drop bookkeeping for files that are gone so the map can't grow unbounded. + for (const key of [...this.orphanFirstSeenAt.keys()]) { + if (!present.has(key)) this.orphanFirstSeenAt.delete(key); + } + if (removed > 0) this.logger.log(`Status media orphan sweep removed ${removed} file(s)`); + return removed; } }