diff --git a/packages/iios-service/prisma/migrations/20260722193958_media_object_tracking/migration.sql b/packages/iios-service/prisma/migrations/20260722193958_media_object_tracking/migration.sql new file mode 100644 index 0000000..6cf15ec --- /dev/null +++ b/packages/iios-service/prisma/migrations/20260722193958_media_object_tracking/migration.sql @@ -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"); diff --git a/packages/iios-service/prisma/schema.prisma b/packages/iios-service/prisma/schema.prisma index 799aeeb..c59a711 100644 --- a/packages/iios-service/prisma/schema.prisma +++ b/packages/iios-service/prisma/schema.prisma @@ -543,6 +543,22 @@ model IiosMessagePart { @@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. model IiosOutboxEvent { eventId String @id @default(cuid()) diff --git a/packages/iios-service/src/media/media.service.spec.ts b/packages/iios-service/src/media/media.service.spec.ts index 8778671..89721e1 100644 --- a/packages/iios-service/src/media/media.service.spec.ts +++ b/packages/iios-service/src/media/media.service.spec.ts @@ -13,9 +13,11 @@ const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/i const prisma = new PrismaClient({ datasources: { db: { url } } }); const asService = prisma as unknown as PrismaService; 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 DAY_MS = 24 * 60 * 60 * 1000; + // point storage at an isolated temp dir for the test run 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(); }); }); + +describe('MediaService orphan sweep', () => { + async function upload(s: MediaService): Promise { + 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(); + }); +}); diff --git a/packages/iios-service/src/media/media.service.ts b/packages/iios-service/src/media/media.service.ts index ad7f8be..eb2cf11 100644 --- a/packages/iios-service/src/media/media.service.ts +++ b/packages/iios-service/src/media/media.service.ts @@ -1,12 +1,15 @@ 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 type { IiosPlatformPorts } from '@insignia/iios-contracts'; import { PLATFORM_PORTS } from '../platform/platform-ports'; +import { PrismaService } from '../prisma/prisma.service'; import { decideOrThrow } from '../platform/fail-closed'; import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver'; 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 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. */ @Injectable() -export class MediaService { +export class MediaService implements OnModuleInit, OnModuleDestroy { 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(/\/$/, ''); + /** 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; constructor( @Inject(STORAGE_PORT) private readonly storage: StoragePort, @Inject(PLATFORM_PORTS) private readonly ports: IiosPlatformPorts, 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. */ async presignUpload( principal: MessagePrincipal, @@ -44,9 +66,58 @@ export class MediaService { const t = this.verify(token, 'put'); 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); + // 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 }; } + /** + * 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 { + 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). */ async presignDownload(principal: MessagePrincipal, contentRef: string, mime = 'application/octet-stream'): Promise<{ url: string }> { const scope = await this.actors.resolveScope(principal); diff --git a/packages/iios-service/src/test-utils/reset-db.ts b/packages/iios-service/src/test-utils/reset-db.ts index 7a01cc0..91a09f9 100644 --- a/packages/iios-service/src/test-utils/reset-db.ts +++ b/packages/iios-service/src/test-utils/reset-db.ts @@ -13,7 +13,7 @@ export async function resetDb(prisma: PrismaClient): Promise { "IiosCalendarSyncCursor","IiosCalendarEvent","IiosCalendarProviderAccount","IiosAvailabilityWindow", "IiosAiEvidenceLink","IiosAiClaim","IiosAiToolCall","IiosAiArtifact","IiosAiModelRun","IiosAiJob","IiosEmbeddingRef", "IiosRouteDecision","IiosRouteBinding","IiosModerationFlag", - "IiosInboundRawEvent","IiosDeliveryAttempt","IiosOutboundCommand","IiosRateLimitBucket", + "IiosInboundRawEvent","IiosDeliveryAttempt","IiosOutboundCommand","IiosRateLimitBucket","IiosMediaObject", "IiosTicketStateHistory","IiosTicketThreadLink","IiosCallbackRequest","IiosTicket", "IiosSupportTeamMember","IiosSupportQueue", "IiosInboxItemStateHistory","IiosInboxItem","IiosUnreadCounter","IiosMessageReceipt",