feat(media): orphan cleanup — track objects + reap unreferenced attachments

Attachments upload direct-to-storage on attach, so an abandoned attach (removed,
cancelled, tab closed) would linger forever. Add IiosMediaObject: a row is recorded
when bytes land (MediaService.put); a scheduled sweepOrphans() reaps objects past a
grace window (IIOS_MEDIA_ORPHAN_GRACE_MS, default 24h) that no IiosMessagePart
references (derived live — no flag to drift). Storage delete via the StoragePort;
best-effort + idempotent. Gated on IIOS_MEDIA_GC_INTERVAL_MS (off by default).
Migration 20260722193958_media_object_tracking. 10 media tests green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-23 01:19:11 +05:30
parent 22eaf0f654
commit 0ee63c139f
5 changed files with 161 additions and 4 deletions
@@ -0,0 +1,20 @@
-- AlterTable
ALTER TABLE "IiosClientRegistry" ALTER COLUMN "allowedAppIds" DROP DEFAULT;
-- CreateTable
CREATE TABLE "IiosMediaObject" (
"id" TEXT NOT NULL,
"scopeId" TEXT NOT NULL,
"objectKey" TEXT NOT NULL,
"mime" TEXT NOT NULL,
"sizeBytes" BIGINT NOT NULL,
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "IiosMediaObject_pkey" PRIMARY KEY ("id")
);
-- CreateIndex
CREATE UNIQUE INDEX "IiosMediaObject_objectKey_key" ON "IiosMediaObject"("objectKey");
-- CreateIndex
CREATE INDEX "IiosMediaObject_createdAt_idx" ON "IiosMediaObject"("createdAt");
@@ -543,6 +543,22 @@ model IiosMessagePart {
@@unique([interactionId, partIndex]) @@unique([interactionId, partIndex])
} }
/// A stored media object (bytes behind the StoragePort). A row is recorded when bytes actually
/// land (MediaService.put), so an orphan sweep can reap objects that no message part references —
/// e.g. a file attached in a composer but never sent. "Referenced" is derived at sweep time from
/// IiosMessagePart.contentRef (no stored flag to drift), after a grace period so in-progress
/// composes are never reaped.
model IiosMediaObject {
id String @id @default(cuid())
scopeId String
objectKey String @unique
mime String
sizeBytes BigInt
createdAt DateTime @default(now())
@@index([createdAt])
}
/// Transactional outbox — written in the same tx as the business row. /// Transactional outbox — written in the same tx as the business row.
model IiosOutboxEvent { model IiosOutboxEvent {
eventId String @id @default(cuid()) eventId String @id @default(cuid())
@@ -13,9 +13,11 @@ const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/i
const prisma = new PrismaClient({ datasources: { db: { url } } }); const prisma = new PrismaClient({ datasources: { db: { url } } });
const asService = prisma as unknown as PrismaService; const asService = prisma as unknown as PrismaService;
const ports = { ...makeFakePorts(), opa: new DevOpaPort() } as IiosPlatformPorts; const ports = { ...makeFakePorts(), opa: new DevOpaPort() } as IiosPlatformPorts;
const svc = () => new MediaService(new LocalDiskStorage(), ports, new ActorResolver(asService)); const svc = () => new MediaService(new LocalDiskStorage(), ports, new ActorResolver(asService), asService);
const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Alice' }; const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Alice' };
const DAY_MS = 24 * 60 * 60 * 1000;
// point storage at an isolated temp dir for the test run // point storage at an isolated temp dir for the test run
process.env.MEDIA_DIR = process.env.MEDIA_DIR ?? '/tmp/iios-media-test'; process.env.MEDIA_DIR = process.env.MEDIA_DIR ?? '/tmp/iios-media-test';
@@ -70,3 +72,51 @@ describe('MediaService (presigned local storage)', () => {
await expect(svc().presignDownload(alice, 'some-other-scope/abc', 'image/png')).rejects.toThrow(); await expect(svc().presignDownload(alice, 'some-other-scope/abc', 'image/png')).rejects.toThrow();
}); });
}); });
describe('MediaService orphan sweep', () => {
async function upload(s: MediaService): Promise<string> {
const bytes = Buffer.from('orphan-candidate');
const { objectKey, uploadUrl } = await s.presignUpload(alice, { mime: 'image/png', sizeBytes: bytes.length });
await s.put(tokenFrom(uploadUrl), bytes);
return objectKey;
}
it('put() records a tracking row for the stored object', async () => {
const key = await upload(svc());
const row = await prisma.iiosMediaObject.findUnique({ where: { objectKey: key } });
expect(row).not.toBeNull();
expect(row!.mime).toBe('image/png');
});
it('reaps an object that is past the grace window and unreferenced', async () => {
const s = svc();
const key = await upload(s);
// Simulate the grace window having elapsed by sweeping "in the future".
const deleted = await s.sweepOrphans(Date.now() + 2 * DAY_MS);
expect(deleted).toBe(1);
expect(await prisma.iiosMediaObject.findUnique({ where: { objectKey: key } })).toBeNull();
});
it('keeps an object still within the grace window', async () => {
const s = svc();
const key = await upload(s);
expect(await s.sweepOrphans(Date.now())).toBe(0);
expect(await prisma.iiosMediaObject.findUnique({ where: { objectKey: key } })).not.toBeNull();
});
it('keeps an object a message part references, even past grace', async () => {
const s = svc();
const key = await upload(s);
const scope = await new ActorResolver(asService).resolveScope(alice);
await prisma.iiosInteraction.create({
data: {
scopeId: scope.id,
kind: 'MESSAGE',
idempotencyKey: 'ref-1',
parts: { create: [{ partIndex: 0, kind: 'MEDIA_REF', contentRef: key }] },
},
});
expect(await s.sweepOrphans(Date.now() + 2 * DAY_MS)).toBe(0);
expect(await prisma.iiosMediaObject.findUnique({ where: { objectKey: key } })).not.toBeNull();
});
});
@@ -1,12 +1,15 @@
import { randomUUID } from 'node:crypto'; import { randomUUID } from 'node:crypto';
import { BadRequestException, ForbiddenException, Inject, Injectable, PayloadTooLargeException } from '@nestjs/common'; import { BadRequestException, ForbiddenException, Inject, Injectable, Logger, type OnModuleDestroy, type OnModuleInit, PayloadTooLargeException } from '@nestjs/common';
import jwt from 'jsonwebtoken'; import jwt from 'jsonwebtoken';
import type { IiosPlatformPorts } from '@insignia/iios-contracts'; import type { IiosPlatformPorts } from '@insignia/iios-contracts';
import { PLATFORM_PORTS } from '../platform/platform-ports'; import { PLATFORM_PORTS } from '../platform/platform-ports';
import { PrismaService } from '../prisma/prisma.service';
import { decideOrThrow } from '../platform/fail-closed'; import { decideOrThrow } from '../platform/fail-closed';
import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver'; import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver';
import { STORAGE_PORT, type StoragePort } from './storage.port'; import { STORAGE_PORT, type StoragePort } from './storage.port';
const DAY_MS = 24 * 60 * 60 * 1000;
interface UploadToken { op: 'put'; objectKey: string; mime: string; maxBytes: number } interface UploadToken { op: 'put'; objectKey: string; mime: string; maxBytes: number }
interface DownloadToken { op: 'get'; objectKey: string; mime: string } interface DownloadToken { op: 'get'; objectKey: string; mime: string }
@@ -17,16 +20,35 @@ interface DownloadToken { op: 'get'; objectKey: string; mime: string }
* kernel only ever stores the object key (contentRef) on a MessagePart. * kernel only ever stores the object key (contentRef) on a MessagePart.
*/ */
@Injectable() @Injectable()
export class MediaService { export class MediaService implements OnModuleInit, OnModuleDestroy {
private readonly secret = process.env.MEDIA_SECRET?.trim() || 'dev-media-secret'; private readonly secret = process.env.MEDIA_SECRET?.trim() || 'dev-media-secret';
private readonly publicUrl = (process.env.PUBLIC_URL?.trim() || `http://localhost:${process.env.PORT ?? 3200}`).replace(/\/$/, ''); private readonly publicUrl = (process.env.PUBLIC_URL?.trim() || `http://localhost:${process.env.PORT ?? 3200}`).replace(/\/$/, '');
/** Objects unreferenced by any message part AND older than this are reaped by the sweep. The grace
* window must exceed the longest realistic compose time so an attached-but-unsent file survives. */
private readonly orphanGraceMs = Number(process.env.IIOS_MEDIA_ORPHAN_GRACE_MS ?? DAY_MS);
private readonly logger = new Logger(MediaService.name);
private gcTimer?: ReturnType<typeof setInterval>;
constructor( constructor(
@Inject(STORAGE_PORT) private readonly storage: StoragePort, @Inject(STORAGE_PORT) private readonly storage: StoragePort,
@Inject(PLATFORM_PORTS) private readonly ports: IiosPlatformPorts, @Inject(PLATFORM_PORTS) private readonly ports: IiosPlatformPorts,
private readonly actors: ActorResolver, private readonly actors: ActorResolver,
private readonly prisma: PrismaService,
) {} ) {}
onModuleInit(): void {
const ms = Number(process.env.IIOS_MEDIA_GC_INTERVAL_MS ?? 0);
if (ms > 0) {
this.gcTimer = setInterval(() => {
void this.sweepOrphans().catch((err) => this.logger.warn(`media orphan sweep failed: ${(err as Error).message}`));
}, ms);
}
}
onModuleDestroy(): void {
if (this.gcTimer) clearInterval(this.gcTimer);
}
/** Authorize an upload and return a short-lived signed PUT url + the object key. */ /** Authorize an upload and return a short-lived signed PUT url + the object key. */
async presignUpload( async presignUpload(
principal: MessagePrincipal, principal: MessagePrincipal,
@@ -44,9 +66,58 @@ export class MediaService {
const t = this.verify<UploadToken>(token, 'put'); const t = this.verify<UploadToken>(token, 'put');
if (data.length > t.maxBytes) throw new PayloadTooLargeException('upload exceeds the presigned size'); if (data.length > t.maxBytes) throw new PayloadTooLargeException('upload exceeds the presigned size');
const { sizeBytes, checksumSha256 } = await this.storage.put(t.objectKey, data, t.mime); const { sizeBytes, checksumSha256 } = await this.storage.put(t.objectKey, data, t.mime);
// Track the object so the orphan sweep can reap it if it's never referenced by a message part.
// Best-effort: tracking must never fail an otherwise-successful upload. scopeId is the key prefix.
const scopeId = t.objectKey.split('/')[0] ?? 'unknown';
await this.prisma.iiosMediaObject
.upsert({
where: { objectKey: t.objectKey },
create: { scopeId, objectKey: t.objectKey, mime: t.mime, sizeBytes: BigInt(sizeBytes) },
update: { sizeBytes: BigInt(sizeBytes) },
})
.catch((err) => this.logger.warn(`media tracking upsert failed for ${t.objectKey}: ${(err as Error).message}`));
return { objectKey: t.objectKey, sizeBytes, checksumSha256 }; return { objectKey: t.objectKey, sizeBytes, checksumSha256 };
} }
/**
* Reap orphaned media: objects older than the grace window that NO message part references. A
* file attached in a composer uploads immediately (direct-to-storage), so an abandoned attach
* (removed, cancelled, tab closed) would otherwise linger forever — there is no draft/commit step.
* "Referenced" is derived live from IiosMessagePart.contentRef, so nothing can drift. Returns the
* number of objects deleted. Safe to call repeatedly (idempotent); storage delete is best-effort.
*/
async sweepOrphans(now: number = Date.now(), batch = 1000): Promise<number> {
const cutoff = new Date(now - this.orphanGraceMs);
const candidates = await this.prisma.iiosMediaObject.findMany({
where: { createdAt: { lt: cutoff } },
select: { id: true, objectKey: true },
take: batch,
});
if (candidates.length === 0) return 0;
const keys = candidates.map((c) => c.objectKey);
const referenced = new Set(
(
await this.prisma.iiosMessagePart.findMany({
where: { contentRef: { in: keys } },
select: { contentRef: true },
})
)
.map((p) => p.contentRef)
.filter((ref): ref is string => ref !== null),
);
const orphans = candidates.filter((c) => !referenced.has(c.objectKey));
let deleted = 0;
for (const o of orphans) {
await this.storage.remove(o.objectKey).catch((err) => this.logger.warn(`storage remove failed for ${o.objectKey}: ${(err as Error).message}`));
await this.prisma.iiosMediaObject.delete({ where: { id: o.id } }).catch(() => undefined);
deleted += 1;
}
if (deleted > 0) this.logger.log(`media orphan sweep reaped ${deleted} object(s)`);
return deleted;
}
/** Authorize a download and return a short-lived signed GET url (tenant-fenced). */ /** Authorize a download and return a short-lived signed GET url (tenant-fenced). */
async presignDownload(principal: MessagePrincipal, contentRef: string, mime = 'application/octet-stream'): Promise<{ url: string }> { async presignDownload(principal: MessagePrincipal, contentRef: string, mime = 'application/octet-stream'): Promise<{ url: string }> {
const scope = await this.actors.resolveScope(principal); const scope = await this.actors.resolveScope(principal);
@@ -13,7 +13,7 @@ export async function resetDb(prisma: PrismaClient): Promise<void> {
"IiosCalendarSyncCursor","IiosCalendarEvent","IiosCalendarProviderAccount","IiosAvailabilityWindow", "IiosCalendarSyncCursor","IiosCalendarEvent","IiosCalendarProviderAccount","IiosAvailabilityWindow",
"IiosAiEvidenceLink","IiosAiClaim","IiosAiToolCall","IiosAiArtifact","IiosAiModelRun","IiosAiJob","IiosEmbeddingRef", "IiosAiEvidenceLink","IiosAiClaim","IiosAiToolCall","IiosAiArtifact","IiosAiModelRun","IiosAiJob","IiosEmbeddingRef",
"IiosRouteDecision","IiosRouteBinding","IiosModerationFlag", "IiosRouteDecision","IiosRouteBinding","IiosModerationFlag",
"IiosInboundRawEvent","IiosDeliveryAttempt","IiosOutboundCommand","IiosRateLimitBucket", "IiosInboundRawEvent","IiosDeliveryAttempt","IiosOutboundCommand","IiosRateLimitBucket","IiosMediaObject",
"IiosTicketStateHistory","IiosTicketThreadLink","IiosCallbackRequest","IiosTicket", "IiosTicketStateHistory","IiosTicketThreadLink","IiosCallbackRequest","IiosTicket",
"IiosSupportTeamMember","IiosSupportQueue", "IiosSupportTeamMember","IiosSupportQueue",
"IiosInboxItemStateHistory","IiosInboxItem","IiosUnreadCounter","IiosMessageReceipt", "IiosInboxItemStateHistory","IiosInboxItem","IiosUnreadCounter","IiosMessageReceipt",