maaz519/iios
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(service): P3.1 inbox tables + extract ActorResolver

Browse Source 
IiosInboxItem + IiosInboxItemStateHistory (migration inbox-init). Extracted
resolveScope/resolveActor/ensureParticipant into a shared ActorResolver
(IdentityModule) so message + inbox layers share identity resolution; MessageService
delegates (P2 tests unchanged).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Maaz Ahamed
2026-07-01 02:41:50 +05:30
parent 4d05f7c1b1
commit 7b174e24fa
7 changed files with 214 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/iios-service/prisma/migrations/20260630210848_inbox_init/migration.sql
+62
View File
@@ -0,0 +1,62 @@
-- CreateEnum
CREATE TYPE "IiosInboxItemKind" AS ENUM ('NEEDS_REPLY', 'NEEDS_REVIEW', 'NEEDS_APPROVAL', 'SUPPORT_UPDATE', 'MEETING_FOLLOWUP', 'DIGEST', 'SYSTEM_ALERT', 'CRM_OWNER_INTEREST');
-- CreateEnum
CREATE TYPE "IiosInboxState" AS ENUM ('OPEN', 'SNOOZED', 'DONE', 'ARCHIVED', 'CANCELLED', 'STALE');
-- CreateTable
CREATE TABLE "IiosInboxItem" (
"id" TEXT NOT NULL,
"scopeId" TEXT NOT NULL,
"ownerActorId" TEXT NOT NULL,
"kind" "IiosInboxItemKind" NOT NULL,
"state" "IiosInboxState" NOT NULL DEFAULT 'OPEN',
"title" TEXT NOT NULL,
"summary" TEXT,
"priority" TEXT NOT NULL DEFAULT 'NORMAL',
"dueAt" TIMESTAMP(3),
"sourceInteractionId" TEXT,
"threadId" TEXT,
"traceId" TEXT,
"actionRef" JSONB,
"policyDecisionRef" TEXT,
"createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
"updatedAt" TIMESTAMP(3) NOT NULL,
"metadata" JSONB,
CONSTRAINT "IiosInboxItem_pkey" PRIMARY KEY ("id")
);
-- CreateTable
CREATE TABLE "IiosInboxItemStateHistory" (
"id" TEXT NOT NULL,
"inboxItemId" TEXT NOT NULL,
"fromState" "IiosInboxState",
"toState" "IiosInboxState" NOT NULL,
"actorId" TEXT,
"reasonCode" TEXT,
"at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT "IiosInboxItemStateHistory_pkey" PRIMARY KEY ("id")
);
-- CreateIndex
CREATE INDEX "IiosInboxItem_ownerActorId_state_priority_idx" ON "IiosInboxItem"("ownerActorId", "state", "priority");
-- CreateIndex
CREATE INDEX "IiosInboxItemStateHistory_inboxItemId_idx" ON "IiosInboxItemStateHistory"("inboxItemId");
-- AddForeignKey
ALTER TABLE "IiosInboxItem" ADD CONSTRAINT "IiosInboxItem_scopeId_fkey" FOREIGN KEY ("scopeId") REFERENCES "IiosScope"("id") ON DELETE CASCADE ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "IiosInboxItem" ADD CONSTRAINT "IiosInboxItem_ownerActorId_fkey" FOREIGN KEY ("ownerActorId") REFERENCES "IiosActorRef"("id") ON DELETE RESTRICT ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "IiosInboxItem" ADD CONSTRAINT "IiosInboxItem_sourceInteractionId_fkey" FOREIGN KEY ("sourceInteractionId") REFERENCES "IiosInteraction"("id") ON DELETE SET NULL ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "IiosInboxItem" ADD CONSTRAINT "IiosInboxItem_threadId_fkey" FOREIGN KEY ("threadId") REFERENCES "IiosThread"("id") ON DELETE SET NULL ON UPDATE CASCADE;
-- AddForeignKey
ALTER TABLE "IiosInboxItemStateHistory" ADD CONSTRAINT "IiosInboxItemStateHistory_inboxItemId_fkey" FOREIGN KEY ("inboxItemId") REFERENCES "IiosInboxItem"("id") ON DELETE CASCADE ON UPDATE CASCADE;
packages/iios-service/prisma/schema.prisma
+72 -2
View File
@@ -75,6 +75,26 @@ enum IiosReceiptKind {
RATE_LIMITED
}
enum IiosInboxItemKind {
NEEDS_REPLY
NEEDS_REVIEW
NEEDS_APPROVAL
SUPPORT_UPDATE
MEETING_FOLLOWUP
DIGEST
SYSTEM_ALERT
CRM_OWNER_INTEREST
}
enum IiosInboxState {
OPEN
SNOOZED
DONE
ARCHIVED
CANCELLED
STALE
}
// ─── Kernel tables ────────────────────────────────────────────────
/// Frozen six-vector scope. orgId + appId are REQUIRED (no global-by-null).
@@ -93,6 +113,7 @@ model IiosScope {
channels IiosChannel[]
threads IiosThread[]
interactions IiosInteraction[]
inboxItems IiosInboxItem[]
@@index([orgId, appId, tenantId])
}
@@ -135,6 +156,7 @@ model IiosActorRef {
interactions IiosInteraction[]
receipts IiosMessageReceipt[]
unreadCounters IiosUnreadCounter[]
inboxItemsOwned IiosInboxItem[]
}
/// A configured channel surface (PORTAL in P1).
@@ -177,6 +199,7 @@ model IiosThread {
participants IiosThreadParticipant[]
interactions IiosInteraction[]
unreadCounters IiosUnreadCounter[]
inboxItems IiosInboxItem[]
@@index([scopeId, status])
}
@@ -219,8 +242,9 @@ model IiosInteraction {
traceId String?
metadata Json?
parts IiosMessagePart[]
receipts IiosMessageReceipt[]
parts IiosMessagePart[]
receipts IiosMessageReceipt[]
inboxItems IiosInboxItem[]
@@unique([scopeId, idempotencyKey])
@@index([threadId, occurredAt])
@@ -299,3 +323,49 @@ model IiosUnreadCounter {
@@id([threadId, actorId])
}
// ─── P3: Inbox (work/awareness surface) ───────────────────────────
/// A durable action/awareness item owned by one actor. Fed by the event
/// projector (deterministic, collapse-per-thread). NOT a thin unread projection.
model IiosInboxItem {
id String @id @default(cuid())
scopeId String
scope IiosScope @relation(fields: [scopeId], references: [id], onDelete: Cascade)
ownerActorId String
ownerActor IiosActorRef @relation(fields: [ownerActorId], references: [id])
kind IiosInboxItemKind
state IiosInboxState @default(OPEN)
title String
summary String?
priority String @default("NORMAL")
dueAt DateTime?
sourceInteractionId String?
sourceInteraction IiosInteraction? @relation(fields: [sourceInteractionId], references: [id])
threadId String?
thread IiosThread? @relation(fields: [threadId], references: [id])
traceId String?
actionRef Json?
policyDecisionRef String?
createdAt DateTime @default(now())
updatedAt DateTime @updatedAt
metadata Json?
stateHistory IiosInboxItemStateHistory[]
@@index([ownerActorId, state, priority])
}
/// Audit trail of inbox-item state transitions.
model IiosInboxItemStateHistory {
id String @id @default(cuid())
inboxItemId String
inboxItem IiosInboxItem @relation(fields: [inboxItemId], references: [id], onDelete: Cascade)
fromState IiosInboxState?
toState IiosInboxState
actorId String?
reasonCode String?
at DateTime @default(now())
@@index([inboxItemId])
}
packages/iios-service/src/app.module.ts
+2 -1
View File
@@ -1,6 +1,7 @@
import { Module } from '@nestjs/common';
import { PrismaModule } from './prisma/prisma.module';
import { PlatformModule } from './platform/platform.module';
import { IdentityModule } from './identity/identity.module';
import { InteractionsModule } from './interactions/interactions.module';
import { OutboxModule } from './outbox/outbox.module';
import { ThreadsModule } from './threads/threads.module';
@@ -9,7 +10,7 @@ import { HealthController } from './health.controller';
import { DevController } from './dev/dev.controller';
@Module({
imports: [PrismaModule, PlatformModule, InteractionsModule, OutboxModule, ThreadsModule, MessageModule],
imports: [PrismaModule, PlatformModule, IdentityModule, InteractionsModule, OutboxModule, ThreadsModule, MessageModule],
controllers: [HealthController, DevController],
})
export class AppModule {}
packages/iios-service/src/identity/actor.resolver.ts
+55
View File
@@ -0,0 +1,55 @@
import { Injectable } from '@nestjs/common';
import { PrismaService } from '../prisma/prisma.service';
/** The authenticated principal derived from the session port (token). */
export interface MessagePrincipal {
userId: string;
orgId: string;
appId: string;
tenantId?: string;
displayName?: string;
}
/**
* Resolves a principal to kernel rows (scope → source handle → actor) and
* manages thread participation. Shared by the message + inbox layers so neither
* re-implements identity resolution. Unknown handles stay UNVERIFIED (the MDM
* port resolution is applied at ingest; native users get a PORTAL_USER handle).
*/
@Injectable()
export class ActorResolver {
constructor(private readonly prisma: PrismaService) {}
async resolveScope(principal: MessagePrincipal) {
return (
(await this.prisma.iiosScope.findFirst({
where: { orgId: principal.orgId, appId: principal.appId, tenantId: principal.tenantId ?? null },
})) ??
(await this.prisma.iiosScope.create({
data: { orgId: principal.orgId, appId: principal.appId, tenantId: principal.tenantId },
}))
);
}
async resolveActor(scopeId: string, principal: MessagePrincipal) {
const handle = await this.prisma.iiosSourceHandle.upsert({
where: { scopeId_kind_externalId: { scopeId, kind: 'PORTAL_USER', externalId: principal.userId } },
create: { scopeId, kind: 'PORTAL_USER', externalId: principal.userId, displayName: principal.displayName },
update: { lastSeenAt: new Date(), displayName: principal.displayName },
});
return (
(await this.prisma.iiosActorRef.findFirst({ where: { sourceHandleId: handle.id } })) ??
(await this.prisma.iiosActorRef.create({
data: { kind: 'HUMAN', sourceHandleId: handle.id, displayName: principal.displayName },
}))
);
}
async ensureParticipant(threadId: string, actorId: string): Promise<void> {
await this.prisma.iiosThreadParticipant.upsert({
where: { threadId_actorId: { threadId, actorId } },
create: { threadId, actorId },
update: {},
});
}
}
packages/iios-service/src/identity/identity.module.ts
+9
View File
@@ -0,0 +1,9 @@
import { Global, Module } from '@nestjs/common';
import { ActorResolver } from './actor.resolver';
@Global()
@Module({
providers: [ActorResolver],
exports: [ActorResolver],
})
export class IdentityModule {}
packages/iios-service/src/messaging/message.service.ts
+12 -50
View File
@@ -5,15 +5,9 @@ import { CloudEvent, IIOS_EVENTS, IiosPlatformPorts } from '@insignia/iios-contr
import { PrismaService } from '../prisma/prisma.service';
import { PLATFORM_PORTS } from '../platform/platform-ports';
import { decideOrThrow } from '../platform/fail-closed';
import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver';
/** The authenticated sender, derived from the session port (token). */
export interface MessagePrincipal {
userId: string;
orgId: string;
appId: string;
tenantId?: string;
displayName?: string;
}
export type { MessagePrincipal };
export interface MessageDto {
id: string;
@@ -42,26 +36,27 @@ export class MessageService {
constructor(
private readonly prisma: PrismaService,
@Inject(PLATFORM_PORTS) private readonly ports: IiosPlatformPorts,
private readonly actors: ActorResolver,
) {}
/** Open an existing thread, or create one when no id is given. Joins as participant. */
async openThread(threadId: string | null, principal: MessagePrincipal): Promise<OpenThreadResult> {
if (!threadId) {
await decideOrThrow(this.ports, { action: 'iios.thread.create', scope: principal });
const scope = await this.resolveScope(principal);
const actor = await this.resolveActor(scope.id, principal);
const scope = await this.actors.resolveScope(principal);
const actor = await this.actors.resolveActor(scope.id, principal);
const thread = await this.prisma.iiosThread.create({
data: { scopeId: scope.id, createdByActorId: actor.id },
});
await this.ensureParticipant(thread.id, actor.id);
await this.actors.ensureParticipant(thread.id, actor.id);
return { threadId: thread.id, status: thread.status, history: [] };
}
const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
if (!thread) throw new NotFoundException('thread not found');
await decideOrThrow(this.ports, { action: 'iios.thread.read', threadId, scopeId: thread.scopeId });
const actor = await this.resolveActor(thread.scopeId, principal);
await this.ensureParticipant(threadId, actor.id);
const actor = await this.actors.resolveActor(thread.scopeId, principal);
await this.actors.ensureParticipant(threadId, actor.id);
return { threadId, status: thread.status, history: await this.history(threadId) };
}
@@ -77,8 +72,8 @@ export class MessageService {
await decideOrThrow(this.ports, { action: 'iios.message.send', threadId, scopeId: thread.scopeId });
const actor = await this.resolveActor(thread.scopeId, principal);
await this.ensureParticipant(threadId, actor.id);
const actor = await this.actors.resolveActor(thread.scopeId, principal);
await this.actors.ensureParticipant(threadId, actor.id);
// Idempotency: a repeat key returns the existing message (no re-increment).
const existing = await this.prisma.iiosInteraction.findUnique({
@@ -168,7 +163,7 @@ export class MessageService {
): Promise<{ interactionId: string; actorId: string }> {
const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
if (!thread) throw new NotFoundException('thread not found');
const actor = await this.resolveActor(thread.scopeId, principal);
const actor = await this.actors.resolveActor(thread.scopeId, principal);
await this.prisma.iiosMessageReceipt.upsert({
where: { interactionId_actorId_receiptKind: { interactionId, actorId: actor.id, receiptKind: 'READ' } },
@@ -190,7 +185,7 @@ export class MessageService {
): Promise<{ interactionId: string; actorId: string }> {
const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
if (!thread) throw new NotFoundException('thread not found');
const actor = await this.resolveActor(thread.scopeId, principal);
const actor = await this.actors.resolveActor(thread.scopeId, principal);
await this.prisma.iiosMessageReceipt.upsert({
where: { interactionId_actorId_receiptKind: { interactionId, actorId: actor.id, receiptKind: 'DELIVERED' } },
create: { interactionId, actorId: actor.id, receiptKind: 'DELIVERED' },
@@ -219,39 +214,6 @@ export class MessageService {
// ─── helpers ────────────────────────────────────────────────────
private async resolveScope(principal: MessagePrincipal) {
return (
(await this.prisma.iiosScope.findFirst({
where: { orgId: principal.orgId, appId: principal.appId, tenantId: principal.tenantId ?? null },
})) ??
(await this.prisma.iiosScope.create({
data: { orgId: principal.orgId, appId: principal.appId, tenantId: principal.tenantId },
}))
);
}
private async resolveActor(scopeId: string, principal: MessagePrincipal) {
const handle = await this.prisma.iiosSourceHandle.upsert({
where: { scopeId_kind_externalId: { scopeId, kind: 'PORTAL_USER', externalId: principal.userId } },
create: { scopeId, kind: 'PORTAL_USER', externalId: principal.userId, displayName: principal.displayName },
update: { lastSeenAt: new Date(), displayName: principal.displayName },
});
return (
(await this.prisma.iiosActorRef.findFirst({ where: { sourceHandleId: handle.id } })) ??
(await this.prisma.iiosActorRef.create({
data: { kind: 'HUMAN', sourceHandleId: handle.id, displayName: principal.displayName },
}))
);
}
private async ensureParticipant(threadId: string, actorId: string): Promise<void> {
await this.prisma.iiosThreadParticipant.upsert({
where: { threadId_actorId: { threadId, actorId } },
create: { threadId, actorId },
update: {},
});
}
private toDto(
interaction: { id: string; actorId: string | null; traceId: string | null; occurredAt: Date; parts: Array<{ kind: string; bodyText: string | null; contentRef: string | null }> },
threadId: string,
packages/iios-service/src/messaging/message.spec.ts
+2 -1
View File
@@ -2,12 +2,13 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
import { PrismaClient } from '@prisma/client';
import { makeFakePorts } from '@insignia/iios-testkit';
import { MessageService, type MessagePrincipal } from './message.service';
import { ActorResolver } from '../identity/actor.resolver';
import type { PrismaService } from '../prisma/prisma.service';
const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/iios?schema=public';
const prisma = new PrismaClient({ datasources: { db: { url } } });
const asService = prisma as unknown as PrismaService;
const svc = () => new MessageService(asService, makeFakePorts());
const svc = () => new MessageService(asService, makeFakePorts(), new ActorResolver(asService));
const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Alice' };
const bob: MessagePrincipal = { userId: 'bob', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Bob' };