feat(iios): @mentions via inbox fan-out + my-annotations query (pins/saves)

Mentions ride the existing event-driven inbox — no chat parsing in the kernel:
- send() carries an OPAQUE mentions[] (userIds) into the message event; the kernel
  never parses "@". The app supplies the notify-list.
- InboxProjector fans out a MENTION inbox item (new generic inbox kind) to each
  mentioned *participant* (never the sender), idempotent per source message; reading
  the thread resolves the reader's MENTION + NEEDS_REPLY items to DONE.
- New MENTION value in the generic IiosInboxItemKind taxonomy (migration).

Pins/saves reuse the annotation primitive; new generic query powers a Saved list:
- MessageService.listMyAnnotated(principal, type) + GET /v1/threads/my-annotations
  returns the caller's annotated messages (type "save") with thread context.

Tests: 182 pass (+4: mention fan-out, resolve-on-read, saved query, opaque mentions).
Verified live: mention→inbox delivery, read→resolve, pin/save persistence.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-07 01:35:30 +05:30
parent f3c4ba72b5
commit 8c814d9b86
8 changed files with 181 additions and 11 deletions
@@ -0,0 +1,2 @@
-- AlterEnum
ALTER TYPE "IiosInboxItemKind" ADD VALUE 'MENTION';
@@ -79,6 +79,7 @@ enum IiosInboxItemKind {
NEEDS_REPLY NEEDS_REPLY
NEEDS_REVIEW NEEDS_REVIEW
NEEDS_APPROVAL NEEDS_APPROVAL
MENTION
SUPPORT_UPDATE SUPPORT_UPDATE
MEETING_FOLLOWUP MEETING_FOLLOWUP
DIGEST DIGEST
@@ -10,6 +10,7 @@ interface MessageSentData {
interactionId: string; interactionId: string;
threadId: string; threadId: string;
senderActorId: string; senderActorId: string;
mentions?: string[]; // opaque app notify-list (userIds); the kernel never parses "@"
} }
interface MessageReadData { interface MessageReadData {
threadId: string; threadId: string;
@@ -57,6 +58,60 @@ export class InboxProjector implements OnModuleInit {
if (p.actorId === data.senderActorId) continue; if (p.actorId === data.senderActorId) continue;
await this.upsertNeedsReply(thread.scopeId, p.actorId, data.threadId, data.interactionId, traceId, thread.subject); await this.upsertNeedsReply(thread.scopeId, p.actorId, data.threadId, data.interactionId, traceId, thread.subject);
} }
// Targeted mention notifications from the app-supplied opaque notify-list.
const mentions = data.mentions ?? [];
if (mentions.length > 0) {
const src = await this.prisma.iiosInteraction.findUnique({
where: { id: data.interactionId },
include: { parts: { where: { kind: 'TEXT' }, take: 1 }, actor: { include: { sourceHandle: true } } },
});
const senderName = src?.actor?.sourceHandle?.externalId ?? 'Someone';
const text = src?.parts[0]?.bodyText ?? undefined;
for (const userId of mentions) {
await this.createMention(thread.scopeId, userId, data.threadId, data.interactionId, data.senderActorId, senderName, text, traceId);
}
}
}
/** One MENTION item per (owner, source message) for a mentioned participant (never the sender). */
private async createMention(
scopeId: string,
userId: string,
threadId: string,
sourceInteractionId: string,
senderActorId: string,
senderName: string,
summary: string | undefined,
traceId: string | undefined,
): Promise<void> {
const handle = await this.prisma.iiosSourceHandle.findUnique({
where: { scopeId_kind_externalId: { scopeId, kind: 'PORTAL_USER', externalId: userId } },
});
if (!handle) return;
const actor = await this.prisma.iiosActorRef.findFirst({ where: { sourceHandleId: handle.id } });
if (!actor || actor.id === senderActorId) return; // resolvable, and never notify the sender
const isMember = await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: actor.id } } });
if (!isMember) return; // only thread participants get mention items
const existing = await this.prisma.iiosInboxItem.findFirst({ where: { ownerActorId: actor.id, sourceInteractionId, kind: 'MENTION' } });
if (existing) return; // idempotent per source message
const item = await this.prisma.iiosInboxItem.create({
data: {
scopeId,
ownerActorId: actor.id,
kind: 'MENTION',
title: `${senderName} mentioned you`,
summary,
priority: 'HIGH',
threadId,
sourceInteractionId,
traceId,
},
});
await this.prisma.iiosInboxItemStateHistory.create({
data: { inboxItemId: item.id, toState: 'OPEN', reasonCode: 'mentioned' },
});
} }
async onMessageRead(event: CloudEvent): Promise<void> { async onMessageRead(event: CloudEvent): Promise<void> {
@@ -67,21 +122,24 @@ export class InboxProjector implements OnModuleInit {
private async applyMessageRead(event: CloudEvent): Promise<void> { private async applyMessageRead(event: CloudEvent): Promise<void> {
const data = event.data as MessageReadData; const data = event.data as MessageReadData;
const item = await this.prisma.iiosInboxItem.findFirst({ // Reading the thread clears the reader's activity AND mention items for it.
const items = await this.prisma.iiosInboxItem.findMany({
where: { where: {
threadId: data.threadId, threadId: data.threadId,
ownerActorId: data.actorId, ownerActorId: data.actorId,
kind: 'NEEDS_REPLY', kind: { in: ['NEEDS_REPLY', 'MENTION'] },
state: { in: ['OPEN', 'SNOOZED'] }, state: { in: ['OPEN', 'SNOOZED'] },
}, },
}); });
if (!item) return; if (items.length === 0) return;
await this.prisma.$transaction([ await this.prisma.$transaction(
this.prisma.iiosInboxItem.update({ where: { id: item.id }, data: { state: 'DONE' } }), items.flatMap((item) => [
this.prisma.iiosInboxItemStateHistory.create({ this.prisma.iiosInboxItem.update({ where: { id: item.id }, data: { state: 'DONE' } }),
data: { inboxItemId: item.id, fromState: item.state, toState: 'DONE', reasonCode: 'message_read' }, this.prisma.iiosInboxItemStateHistory.create({
}), data: { inboxItemId: item.id, fromState: item.state, toState: 'DONE', reasonCode: 'message_read' },
]); }),
]),
);
} }
private async upsertNeedsReply( private async upsertNeedsReply(
@@ -88,6 +88,38 @@ describe('InboxProjector (P3 work surface)', () => {
expect(await itemsFor('bob')).toHaveLength(1); expect(await itemsFor('bob')).toHaveLength(1);
}); });
it('message.sent with mentions → a MENTION item for the mentioned participant only', async () => {
const m = ms();
const { threadId } = await m.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
await m.addParticipant(threadId, alice, 'bob');
await m.addParticipant(threadId, alice, 'carol');
await m.send(threadId, alice, { content: 'hey @bob look' }, 'k1', undefined, undefined, ['bob']);
await projector().onMessageSent((await eventsOf(IIOS_EVENTS.messageSent))[0]!);
const bobMentions = (await itemsFor('bob')).filter((i) => i.kind === 'MENTION');
expect(bobMentions).toHaveLength(1);
expect(bobMentions[0]?.state).toBe('OPEN');
expect(bobMentions[0]?.sourceInteractionId).toBeTruthy();
expect((await itemsFor('carol')).filter((i) => i.kind === 'MENTION')).toHaveLength(0); // not mentioned
expect((await itemsFor('alice')).filter((i) => i.kind === 'MENTION')).toHaveLength(0); // never the sender
});
it('reading the thread resolves the readers MENTION item to DONE', async () => {
const m = ms();
const { threadId } = await m.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
await m.addParticipant(threadId, alice, 'bob');
const sent = await m.send(threadId, alice, { content: '@bob ping' }, 'k1', undefined, undefined, ['bob']);
const proj = projector();
await proj.onMessageSent((await eventsOf(IIOS_EVENTS.messageSent))[0]!);
expect((await itemsFor('bob')).filter((i) => i.kind === 'MENTION')[0]?.state).toBe('OPEN');
await m.markRead(threadId, bob, sent.id);
await proj.onMessageRead((await eventsOf(IIOS_EVENTS.messageRead))[0]!);
expect((await itemsFor('bob')).filter((i) => i.kind === 'MENTION')[0]?.state).toBe('DONE');
});
it('message.read → the recipient item goes DONE', async () => { it('message.read → the recipient item goes DONE', async () => {
const m = ms(); const m = ms();
const { threadId } = await m.openThread(null, alice); const { threadId } = await m.openThread(null, alice);
@@ -91,7 +91,7 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection {
@SubscribeMessage('send_message') @SubscribeMessage('send_message')
async sendMessage( async sendMessage(
@ConnectedSocket() client: Socket, @ConnectedSocket() client: Socket,
@MessageBody() body: { threadId: string; content: string; contentRef?: string; parentInteractionId?: string }, @MessageBody() body: { threadId: string; content: string; contentRef?: string; parentInteractionId?: string; mentions?: string[] },
) { ) {
const { principal } = client.data as SocketState; const { principal } = client.data as SocketState;
const msg = await this.messages.send( const msg = await this.messages.send(
@@ -101,6 +101,7 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection {
randomUUID(), randomUUID(),
undefined, undefined,
body.parentInteractionId, body.parentInteractionId,
body.mentions,
); );
this.emittedLocally.add(msg.id); this.emittedLocally.add(msg.id);
this.server.to(body.threadId).emit('message', msg); this.server.to(body.threadId).emit('message', msg);
@@ -191,6 +191,7 @@ export class MessageService {
idempotencyKey: string, idempotencyKey: string,
traceId: string = randomUUID(), traceId: string = randomUUID(),
parentInteractionId?: string, parentInteractionId?: string,
mentions?: string[],
): Promise<MessageDto> { ): Promise<MessageDto> {
const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } }); const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
if (!thread) throw new NotFoundException('thread not found'); if (!thread) throw new NotFoundException('thread not found');
@@ -245,7 +246,9 @@ export class MessageService {
datacontenttype: 'application/json', datacontenttype: 'application/json',
traceparent: `00-${traceId.replace(/-/g, '')}-0000000000000000-01`, traceparent: `00-${traceId.replace(/-/g, '')}-0000000000000000-01`,
insignia: { scopeSnapshotId: thread.scopeId, correlationId: traceId, idempotencyKey, dataClass: 'internal' }, insignia: { scopeSnapshotId: thread.scopeId, correlationId: traceId, idempotencyKey, dataClass: 'internal' },
data: { interactionId: created.id, threadId, senderActorId: actor.id }, // `mentions` is an OPAQUE app-supplied notify-list (userIds) — the kernel never parses
// "@"; the inbox projector generically fans out a notification to those actors.
data: { interactionId: created.id, threadId, senderActorId: actor.id, mentions: mentions ?? [] },
}; };
await tx.iiosOutboxEvent.create({ await tx.iiosOutboxEvent.create({
data: { data: {
@@ -416,6 +419,48 @@ export class MessageService {
return { interactionId, threadId: target.threadId, type: annotationType, value, op, users }; return { interactionId, threadId: target.threadId, type: annotationType, value, op, users };
} }
/**
* Generic "my annotated interactions" — every message the caller annotated with a given
* type, newest first, with thread context. The app uses type "save" for a personal
* bookmarks list (and could use "pin" for a cross-thread pinned view). Generic: the kernel
* never interprets the type string.
*/
async listMyAnnotated(
principal: MessagePrincipal,
annotationType: string,
): Promise<Array<{ message: MessageDto; threadId: string; threadSubject: string | null }>> {
const scope = await this.actors.findScope(principal);
if (!scope) return [];
const actor = await this.actors.resolveActor(scope.id, principal);
const anns = await this.prisma.iiosInteractionAnnotation.findMany({
where: { actorId: actor.id, annotationType },
orderBy: { createdAt: 'desc' },
select: { targetInteractionId: true },
});
const ids = anns.map((a) => a.targetInteractionId);
if (ids.length === 0) return [];
const [interactions, agg] = await Promise.all([
this.prisma.iiosInteraction.findMany({
where: { id: { in: ids } },
include: {
parts: { orderBy: { partIndex: 'asc' } },
actor: { include: { sourceHandle: true } },
thread: { select: { subject: true } },
},
}),
this.annotationsByInteraction(ids),
]);
const byId = new Map(interactions.map((i) => [i.id, i]));
return ids
.map((id) => byId.get(id))
.filter((i): i is NonNullable<typeof i> => Boolean(i && i.threadId))
.map((i) => ({
message: this.toDto(i, i.threadId as string, agg.get(i.id) ?? []),
threadId: i.threadId as string,
threadSubject: i.thread?.subject ?? null,
}));
}
// ─── helpers ──────────────────────────────────────────────────── // ─── helpers ────────────────────────────────────────────────────
/** Aggregate annotations for the given interactions into { type, value, users[] } groups. */ /** Aggregate annotations for the given interactions into { type, value, users[] } groups. */
@@ -202,4 +202,28 @@ describe('Interaction annotations (generic reactions primitive)', () => {
const msg = await s.send(threadId, alice, { content: 'secret' }, 'k1'); const msg = await s.send(threadId, alice, { content: 'secret' }, 'k1');
await expect(s.toggleAnnotation(msg.id, bob, 'reaction', '👍')).rejects.toBeInstanceOf(PolicyDeniedError); await expect(s.toggleAnnotation(msg.id, bob, 'reaction', '👍')).rejects.toBeInstanceOf(PolicyDeniedError);
}); });
it('listMyAnnotated returns the callers saved messages with thread context; others see none', async () => {
const s = gov();
const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN', subject: 'Design' });
await s.addParticipant(threadId, alice, 'bob');
const msg = await s.send(threadId, alice, { content: 'save this' }, 'k1');
await s.toggleAnnotation(msg.id, alice, 'save', ''); // save is a personal annotation
const saved = await s.listMyAnnotated(alice, 'save');
expect(saved).toHaveLength(1);
expect(saved[0]).toMatchObject({ threadId, threadSubject: 'Design' });
expect(saved[0]?.message.content).toBe('save this');
expect(await s.listMyAnnotated(bob, 'save')).toHaveLength(0); // bob saved nothing
});
it('send carries an OPAQUE mentions[] into the message event (kernel never parses @)', async () => {
const s = svc();
const { threadId } = await s.openThread(null, alice);
await s.openThread(threadId, bob);
await s.send(threadId, alice, { content: 'hi @bob' }, 'k1', undefined, undefined, ['bob']);
const ev = await prisma.iiosOutboxEvent.findFirstOrThrow({ where: { eventType: 'com.insignia.iios.message.sent.v1' } });
const ce = ev.cloudEvent as { data: { mentions?: string[] } };
expect(ce.data.mentions).toEqual(['bob']);
});
}); });
@@ -8,6 +8,7 @@ import {
HttpCode, HttpCode,
Param, Param,
Post, Post,
Query,
} from '@nestjs/common'; } from '@nestjs/common';
import { ThreadsService } from './threads.service'; import { ThreadsService } from './threads.service';
import { MessageService } from '../messaging/message.service'; import { MessageService } from '../messaging/message.service';
@@ -28,6 +29,12 @@ export class ThreadsController {
return this.messages.listThreads(this.principal(auth)); return this.messages.listThreads(this.principal(auth));
} }
/** Generic "my annotated messages" (e.g. ?type=save for a personal bookmarks list). */
@Get('my-annotations')
async myAnnotations(@Headers('authorization') auth?: string, @Query('type') type = 'save') {
return this.messages.listMyAnnotated(this.principal(auth), type);
}
/** Create a thread; `membership`/`creatorRole`/`subject` are opaque, app-supplied attributes the kernel stores but never interprets. */ /** Create a thread; `membership`/`creatorRole`/`subject` are opaque, app-supplied attributes the kernel stores but never interprets. */
@Post() @Post()
@HttpCode(201) @HttpCode(201)