maaz519
ebf553eb68
The generic primitives channels need beyond dm/group, kept policy-governed and scope-fenced (kernel never interprets 'channel'/'public'): - discoverThreads(principal, filter): scope-wide browse (NOT membership-scoped) matching an opaque metadata filter, each result flagged joined; governed by iios.thread.discover (scope fence in the query). REST: GET /v1/threads/discover - open self-join for PUBLIC channels: openThread now passes visibility into the join decision; dev-OPA iios.thread.join allows membership=channel+visibility= public (dm/group/private-channel stay invite-only) - leaveThread + iios.thread.leave + REST DELETE /v1/threads/:id/me - tests: public self-join vs private denied, discover joined-flags + group excluded + leave; dev-opa join-public/discover/leave rules (29 green) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
733 lines
34 KiB
TypeScript
733 lines
34 KiB
TypeScript
import { randomUUID } from 'node:crypto';
|
|
import { Inject, Injectable, NotFoundException } from '@nestjs/common';
|
|
import { Prisma } from '@prisma/client';
|
|
import { CloudEvent, IIOS_EVENTS, IiosPlatformPorts } from '@insignia/iios-contracts';
|
|
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';
|
|
|
|
export type { MessagePrincipal };
|
|
|
|
/** A generic annotation aggregate on a message (opaque type/value + who applied it). */
|
|
export interface AnnotationDto {
|
|
type: string;
|
|
value: string;
|
|
users: string[];
|
|
}
|
|
|
|
/** A media attachment on a message (the app renders it by `kind`). */
|
|
export interface AttachmentDto {
|
|
contentRef: string;
|
|
mimeType: string;
|
|
sizeBytes: number;
|
|
kind: 'image' | 'video' | 'audio' | 'file';
|
|
}
|
|
|
|
function mediaKind(mime: string): AttachmentDto['kind'] {
|
|
if (mime.startsWith('image/')) return 'image';
|
|
if (mime.startsWith('video/')) return 'video';
|
|
if (mime.startsWith('audio/')) return 'audio';
|
|
return 'file';
|
|
}
|
|
|
|
export interface MessageDto {
|
|
id: string;
|
|
threadId: string;
|
|
senderActorId: string;
|
|
senderId: string; // the sender's stable externalId (email/username) — reliable "is this mine?"
|
|
senderName: string;
|
|
content: string;
|
|
contentRef?: string;
|
|
attachment?: AttachmentDto;
|
|
parentInteractionId?: string;
|
|
annotations: AnnotationDto[];
|
|
traceId: string;
|
|
createdAt: Date;
|
|
}
|
|
|
|
export interface ThreadSummary {
|
|
threadId: string;
|
|
subject: string | null;
|
|
membership?: string;
|
|
/** The thread's opaque, app-supplied attribute bag — echoed back verbatim; the kernel never interprets it. */
|
|
metadata?: Record<string, unknown> | null;
|
|
participants: string[];
|
|
participantCount: number;
|
|
unread: number;
|
|
muted?: boolean;
|
|
lastMessage?: string;
|
|
lastAt?: Date;
|
|
}
|
|
|
|
export interface OpenThreadResult {
|
|
threadId: string;
|
|
status: string;
|
|
history: MessageDto[];
|
|
}
|
|
|
|
/**
|
|
* Native direct messaging (P2) over the P1 kernel. Pure message semantics — no
|
|
* tickets/agents/SLA. Send writes an interaction + parts + a `message.sent`
|
|
* outbox event in one tx, bumps every other participant's unread, and stamps a
|
|
* trace id (DB → event). Reuses the kernel handle/actor resolution from ingest.
|
|
*/
|
|
@Injectable()
|
|
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. On create, an optional
|
|
* generic `membership` attribute is stamped on the thread's metadata (the app's DM/group
|
|
* hint — the kernel never branches on it; policy does), and the creator joins as ADMIN
|
|
* for a group. Opening an existing thread is a GOVERNED join: only an existing member may
|
|
* re-open it (policy `iios.thread.join`) — new members enter via addParticipant.
|
|
*/
|
|
async openThread(threadId: string | null, principal: MessagePrincipal, opts?: { membership?: string; creatorRole?: string; subject?: string; metadata?: Record<string, unknown> }): Promise<OpenThreadResult> {
|
|
if (!threadId) {
|
|
await decideOrThrow(this.ports, { action: 'iios.thread.create', scope: principal });
|
|
const scope = await this.actors.resolveScope(principal);
|
|
const actor = await this.actors.resolveActor(scope.id, principal);
|
|
// `membership`/`creatorRole`/`subject`/`metadata` are generic, app-supplied thread attributes —
|
|
// the kernel stores/echoes them as an opaque bag but never branches on their meaning (that lives
|
|
// in policy + the app). `membership` is folded into the same bag for back-compat.
|
|
const merged = { ...(opts?.metadata ?? {}), ...(opts?.membership ? { membership: opts.membership } : {}) };
|
|
const thread = await this.prisma.iiosThread.create({
|
|
data: {
|
|
scopeId: scope.id,
|
|
createdByActorId: actor.id,
|
|
subject: opts?.subject?.trim() || undefined,
|
|
metadata: Object.keys(merged).length > 0 ? (merged as Prisma.InputJsonValue) : undefined,
|
|
},
|
|
});
|
|
await this.actors.ensureParticipant(thread.id, actor.id, opts?.creatorRole ?? 'MEMBER');
|
|
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.actors.resolveActor(thread.scopeId, principal);
|
|
const alreadyMember =
|
|
(await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: actor.id } } })) !== null;
|
|
// Join is governed ONLY for threads that opted into a membership model (chat dm/group/channel);
|
|
// support/inbox/generic threads (no membership attr) keep open-join, unchanged. A PUBLIC channel
|
|
// is the one membership type that also allows open self-join (policy reads `visibility`).
|
|
const bag = (thread.metadata as { membership?: string; visibility?: string } | null) ?? {};
|
|
await decideOrThrow(this.ports, {
|
|
action: 'iios.thread.join',
|
|
threadId,
|
|
scopeId: thread.scopeId,
|
|
membership: bag.membership,
|
|
visibility: bag.visibility,
|
|
alreadyMember,
|
|
});
|
|
await this.actors.ensureParticipant(threadId, actor.id);
|
|
return { threadId, status: thread.status, history: await this.history(threadId) };
|
|
}
|
|
|
|
/**
|
|
* Governed thread membership (a member adds another user by userId). Fail-closed via
|
|
* policy: the DM cap / group-admin rule is enforced by OPA (dev stub now, real later);
|
|
* the kernel only supplies generic context and adds the participant. The target's actor
|
|
* is resolved-or-created so you can add someone who hasn't logged in yet.
|
|
*/
|
|
async addParticipant(threadId: string, principal: MessagePrincipal, targetUserId: string, role = 'MEMBER'): Promise<{ threadId: string; participantCount: number }> {
|
|
const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
|
|
if (!thread) throw new NotFoundException('thread not found');
|
|
const caller = await this.actors.resolveActor(thread.scopeId, principal);
|
|
const callerP = await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: caller.id } } });
|
|
const participantCount = await this.prisma.iiosThreadParticipant.count({ where: { threadId } });
|
|
const membership = (thread.metadata as { membership?: string } | null)?.membership;
|
|
|
|
await decideOrThrow(this.ports, {
|
|
action: 'iios.thread.participant.add',
|
|
threadId,
|
|
scopeId: thread.scopeId,
|
|
membership,
|
|
participantCount,
|
|
callerRole: callerP?.participantRole,
|
|
targetUserId,
|
|
role,
|
|
});
|
|
|
|
const scope = await this.actors.resolveScope(principal);
|
|
const target = await this.actors.resolveActor(scope.id, {
|
|
userId: targetUserId,
|
|
appId: principal.appId,
|
|
orgId: principal.orgId,
|
|
tenantId: principal.tenantId,
|
|
displayName: targetUserId,
|
|
});
|
|
await this.actors.ensureParticipant(threadId, target.id, role);
|
|
return { threadId, participantCount: participantCount + 1 };
|
|
}
|
|
|
|
/**
|
|
* Governed thread rename (a generic subject update). Policy decides who may rename — for a
|
|
* membership thread the dev OPA requires the caller be a group ADMIN. The kernel only writes
|
|
* the subject; "group settings" meaning lives in the app + policy, not here.
|
|
*/
|
|
async renameThread(threadId: string, principal: MessagePrincipal, subject: string): Promise<{ threadId: string; subject: string }> {
|
|
const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
|
|
if (!thread) throw new NotFoundException('thread not found');
|
|
const caller = await this.actors.resolveActor(thread.scopeId, principal);
|
|
const callerP = await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: caller.id } } });
|
|
const membership = (thread.metadata as { membership?: string } | null)?.membership;
|
|
|
|
await decideOrThrow(this.ports, {
|
|
action: 'iios.thread.update',
|
|
threadId,
|
|
scopeId: thread.scopeId,
|
|
membership,
|
|
callerRole: callerP?.participantRole,
|
|
});
|
|
|
|
await this.prisma.iiosThread.update({ where: { id: threadId }, data: { subject } });
|
|
return { threadId, subject };
|
|
}
|
|
|
|
/**
|
|
* Governed participant removal. Policy decides who may remove — for a membership thread the dev
|
|
* OPA requires the caller be a group ADMIN. Removing a non-participant is a no-op success.
|
|
*/
|
|
async removeParticipant(threadId: string, principal: MessagePrincipal, targetUserId: string): Promise<{ threadId: string; participantCount: number }> {
|
|
const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
|
|
if (!thread) throw new NotFoundException('thread not found');
|
|
const caller = await this.actors.resolveActor(thread.scopeId, principal);
|
|
const callerP = await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: caller.id } } });
|
|
const membership = (thread.metadata as { membership?: string } | null)?.membership;
|
|
|
|
await decideOrThrow(this.ports, {
|
|
action: 'iios.thread.participant.remove',
|
|
threadId,
|
|
scopeId: thread.scopeId,
|
|
membership,
|
|
callerRole: callerP?.participantRole,
|
|
targetUserId,
|
|
});
|
|
|
|
const scope = await this.actors.resolveScope(principal);
|
|
const target = await this.actors.resolveActor(scope.id, {
|
|
userId: targetUserId, appId: principal.appId, orgId: principal.orgId, tenantId: principal.tenantId, displayName: targetUserId,
|
|
});
|
|
await this.prisma.iiosThreadParticipant.deleteMany({ where: { threadId, actorId: target.id } });
|
|
const participantCount = await this.prisma.iiosThreadParticipant.count({ where: { threadId } });
|
|
return { threadId, participantCount };
|
|
}
|
|
|
|
/** Members of a thread with their role — drives the group settings member list. Read is policy-scoped. */
|
|
async listParticipants(threadId: string, principal: MessagePrincipal): Promise<Array<{ userId: string; displayName: string; role: string }>> {
|
|
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 parts = await this.prisma.iiosThreadParticipant.findMany({
|
|
where: { threadId },
|
|
include: { actor: { include: { sourceHandle: true } } },
|
|
});
|
|
return parts.map((p) => ({
|
|
userId: p.actor?.sourceHandle?.externalId ?? '',
|
|
displayName: p.actor?.displayName ?? p.actor?.sourceHandle?.externalId ?? 'unknown',
|
|
role: p.participantRole ?? 'MEMBER',
|
|
}));
|
|
}
|
|
|
|
/**
|
|
* Scope-wide thread discovery — the ONE generic primitive channels need beyond dm/group.
|
|
* Unlike listThreads (membership-scoped), this returns threads in the caller's scope matching an
|
|
* opaque metadata filter (e.g. {membership:'channel', visibility:'public'}) REGARDLESS of whether
|
|
* the caller is a participant, each flagged `joined`. Governed by policy (fail-closed) + fenced to
|
|
* the caller's own scope. The kernel never interprets the filter keys.
|
|
*/
|
|
async discoverThreads(
|
|
principal: MessagePrincipal,
|
|
filter?: { metadata?: Record<string, string> },
|
|
): Promise<Array<{ threadId: string; subject: string | null; metadata: Record<string, unknown> | null; participantCount: number; joined: boolean }>> {
|
|
const scope = await this.actors.findScope(principal);
|
|
if (!scope) return [];
|
|
await decideOrThrow(this.ports, { action: 'iios.thread.discover', scopeId: scope.id });
|
|
const actor = await this.actors.resolveActor(scope.id, principal);
|
|
|
|
const inScope = await this.prisma.iiosThread.findMany({ where: { scopeId: scope.id } });
|
|
const metaFilter = filter?.metadata;
|
|
const matched = metaFilter
|
|
? inScope.filter((t) => {
|
|
const bag = (t.metadata as Record<string, unknown> | null) ?? {};
|
|
return Object.entries(metaFilter).every(([k, v]) => bag[k] === v);
|
|
})
|
|
: inScope;
|
|
if (matched.length === 0) return [];
|
|
|
|
const ids = matched.map((t) => t.id);
|
|
const parts = await this.prisma.iiosThreadParticipant.groupBy({ by: ['threadId'], where: { threadId: { in: ids } }, _count: { actorId: true } });
|
|
const countBy = new Map(parts.map((p) => [p.threadId, p._count.actorId]));
|
|
const mine = await this.prisma.iiosThreadParticipant.findMany({ where: { threadId: { in: ids }, actorId: actor.id }, select: { threadId: true } });
|
|
const joinedSet = new Set(mine.map((m) => m.threadId));
|
|
|
|
return matched.map((t) => ({
|
|
threadId: t.id,
|
|
subject: t.subject,
|
|
metadata: (t.metadata as Record<string, unknown> | null) ?? null,
|
|
participantCount: countBy.get(t.id) ?? 0,
|
|
joined: joinedSet.has(t.id),
|
|
}));
|
|
}
|
|
|
|
/** Self-leave: remove the caller's own participant row. Governed (a member may always leave). */
|
|
async leaveThread(threadId: string, principal: MessagePrincipal): Promise<{ threadId: string; participantCount: number }> {
|
|
const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
|
|
if (!thread) throw new NotFoundException('thread not found');
|
|
const actor = await this.actors.resolveActor(thread.scopeId, principal);
|
|
const callerP = await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: actor.id } } });
|
|
const membership = (thread.metadata as { membership?: string } | null)?.membership;
|
|
await decideOrThrow(this.ports, { action: 'iios.thread.leave', threadId, scopeId: thread.scopeId, membership, callerRole: callerP?.participantRole });
|
|
await this.prisma.iiosThreadParticipant.deleteMany({ where: { threadId, actorId: actor.id } });
|
|
const participantCount = await this.prisma.iiosThreadParticipant.count({ where: { threadId } });
|
|
return { threadId, participantCount };
|
|
}
|
|
|
|
/** Toggle the caller's per-thread notification mute flag. */
|
|
async muteThread(threadId: string, principal: MessagePrincipal, muted: boolean): Promise<{ threadId: string; muted: boolean }> {
|
|
const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
|
|
if (!thread) throw new NotFoundException('thread not found');
|
|
const actor = await this.actors.resolveActor(thread.scopeId, principal);
|
|
await this.prisma.iiosThreadParticipant.update({
|
|
where: { threadId_actorId: { threadId, actorId: actor.id } },
|
|
data: { muted },
|
|
});
|
|
return { threadId, muted };
|
|
}
|
|
|
|
/**
|
|
* Generic "my threads": every thread the caller participates in, with last message + unread.
|
|
* An optional `filter.metadata` narrows to threads whose opaque attribute bag matches ALL of the
|
|
* given key/values (an equality match on the JSON bag — the kernel does not interpret the keys).
|
|
*/
|
|
async listThreads(principal: MessagePrincipal, filter?: { metadata?: Record<string, string> }): Promise<ThreadSummary[]> {
|
|
const scope = await this.actors.findScope(principal);
|
|
if (!scope) return [];
|
|
const actor = await this.actors.resolveActor(scope.id, principal);
|
|
const memberships = await this.prisma.iiosThreadParticipant.findMany({ where: { actorId: actor.id }, select: { threadId: true, muted: true } });
|
|
const threadIds = memberships.map((m) => m.threadId);
|
|
if (threadIds.length === 0) return [];
|
|
const mutedBy = new Map(memberships.map((m) => [m.threadId, m.muted]));
|
|
|
|
const [allThreads, unreads, allParts] = await Promise.all([
|
|
this.prisma.iiosThread.findMany({ where: { id: { in: threadIds } } }),
|
|
this.prisma.iiosUnreadCounter.findMany({ where: { threadId: { in: threadIds }, actorId: actor.id } }),
|
|
this.prisma.iiosThreadParticipant.findMany({
|
|
where: { threadId: { in: threadIds } },
|
|
include: { actor: { include: { sourceHandle: true } } },
|
|
}),
|
|
]);
|
|
// Opaque equality filter on the metadata bag (every requested key must match).
|
|
const metaFilter = filter?.metadata;
|
|
const threads = metaFilter
|
|
? allThreads.filter((t) => {
|
|
const bag = (t.metadata as Record<string, unknown> | null) ?? {};
|
|
return Object.entries(metaFilter).every(([k, v]) => bag[k] === v);
|
|
})
|
|
: allThreads;
|
|
const unreadBy = new Map(unreads.map((u) => [u.threadId, u.unreadCount]));
|
|
const membersBy = new Map<string, string[]>();
|
|
for (const p of allParts) {
|
|
const name = p.actor?.sourceHandle?.externalId ?? p.actor?.displayName ?? p.actorId;
|
|
membersBy.set(p.threadId, [...(membersBy.get(p.threadId) ?? []), name]);
|
|
}
|
|
|
|
// Latest message per thread in ONE query (Postgres DISTINCT ON) instead of one findFirst
|
|
// per thread — a caller with many threads no longer fans out N parallel queries and
|
|
// exhausts the connection pool.
|
|
const lastRows =
|
|
threads.length === 0
|
|
? []
|
|
: await this.prisma.$queryRaw<Array<{ threadId: string; occurredAt: Date; lastMessage: string | null }>>(Prisma.sql`
|
|
SELECT DISTINCT ON (i."threadId")
|
|
i."threadId" AS "threadId",
|
|
i."occurredAt" AS "occurredAt",
|
|
(SELECT p."bodyText" FROM "IiosMessagePart" p
|
|
WHERE p."interactionId" = i.id AND p.kind::text = 'TEXT'
|
|
ORDER BY p."partIndex" ASC LIMIT 1) AS "lastMessage"
|
|
FROM "IiosInteraction" i
|
|
WHERE i."threadId" IN (${Prisma.join(threads.map((t) => t.id))})
|
|
ORDER BY i."threadId", i."occurredAt" DESC
|
|
`);
|
|
const lastBy = new Map(lastRows.map((r) => [r.threadId, r]));
|
|
|
|
const summaries = threads.map((t) => {
|
|
const last = lastBy.get(t.id);
|
|
const members = membersBy.get(t.id) ?? [];
|
|
return {
|
|
threadId: t.id,
|
|
subject: t.subject,
|
|
membership: (t.metadata as { membership?: string } | null)?.membership,
|
|
metadata: (t.metadata as Record<string, unknown> | null) ?? null,
|
|
participants: members,
|
|
participantCount: members.length,
|
|
unread: unreadBy.get(t.id) ?? 0,
|
|
muted: mutedBy.get(t.id) ?? false,
|
|
lastMessage: last?.lastMessage ?? undefined,
|
|
lastAt: last?.occurredAt,
|
|
};
|
|
});
|
|
return summaries.sort((a, b) => (b.lastAt?.getTime() ?? 0) - (a.lastAt?.getTime() ?? 0));
|
|
}
|
|
|
|
async send(
|
|
threadId: string,
|
|
principal: MessagePrincipal,
|
|
body: { content: string; contentRef?: string; mimeType?: string; sizeBytes?: number; checksumSha256?: string },
|
|
idempotencyKey: string,
|
|
traceId: string = randomUUID(),
|
|
parentInteractionId?: string,
|
|
mentions?: string[],
|
|
): Promise<MessageDto> {
|
|
const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
|
|
if (!thread) throw new NotFoundException('thread not found');
|
|
|
|
await decideOrThrow(this.ports, { action: 'iios.message.send', threadId, scopeId: thread.scopeId });
|
|
|
|
const actor = await this.actors.resolveActor(thread.scopeId, principal);
|
|
await this.actors.ensureParticipant(threadId, actor.id);
|
|
|
|
// A reply links to its parent — but only if the parent is in the same thread (else ignored).
|
|
const parentRef = parentInteractionId
|
|
? (await this.prisma.iiosInteraction.findFirst({ where: { id: parentInteractionId, threadId }, select: { id: true } }))?.id
|
|
: undefined;
|
|
|
|
// Idempotency: a repeat key returns the existing message (no re-increment).
|
|
const existing = await this.prisma.iiosInteraction.findUnique({
|
|
where: { scopeId_idempotencyKey: { scopeId: thread.scopeId, idempotencyKey } },
|
|
include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } },
|
|
});
|
|
if (existing) return this.toDto(existing, threadId);
|
|
|
|
try {
|
|
const interaction = await this.prisma.$transaction(async (tx) => {
|
|
const created = await tx.iiosInteraction.create({
|
|
data: {
|
|
scopeId: thread.scopeId,
|
|
kind: 'MESSAGE',
|
|
threadId,
|
|
actorId: actor.id,
|
|
idempotencyKey,
|
|
status: 'NORMALIZED',
|
|
traceId,
|
|
parentInteractionId: parentRef,
|
|
},
|
|
});
|
|
|
|
const parts: Prisma.IiosMessagePartCreateManyInput[] = [
|
|
{ interactionId: created.id, partIndex: 0, kind: 'TEXT', bodyText: body.content },
|
|
];
|
|
if (body.contentRef) {
|
|
const mime = body.mimeType ?? '';
|
|
// Generic part kind from mime — the kernel stores media as opaque parts.
|
|
const kind = /^(image|video)\//.test(mime) ? 'MEDIA_REF' : /^audio\//.test(mime) ? 'VOICE_REF' : 'FILE_REF';
|
|
parts.push({
|
|
interactionId: created.id,
|
|
partIndex: 1,
|
|
kind,
|
|
contentRef: body.contentRef,
|
|
mimeType: body.mimeType,
|
|
sizeBytes: body.sizeBytes != null ? BigInt(body.sizeBytes) : undefined,
|
|
checksumSha256: body.checksumSha256,
|
|
});
|
|
}
|
|
await tx.iiosMessagePart.createMany({ data: parts });
|
|
|
|
const event: CloudEvent = {
|
|
specversion: '1.0',
|
|
id: `evt_${created.id}`,
|
|
type: IIOS_EVENTS.messageSent,
|
|
source: `iios/message/${thread.scopeId}`,
|
|
subject: `interaction/${created.id}`,
|
|
time: new Date().toISOString(),
|
|
datacontenttype: 'application/json',
|
|
traceparent: `00-${traceId.replace(/-/g, '')}-0000000000000000-01`,
|
|
insignia: { scopeSnapshotId: thread.scopeId, correlationId: traceId, idempotencyKey, dataClass: 'internal' },
|
|
// `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({
|
|
data: {
|
|
aggregateType: 'interaction',
|
|
aggregateId: created.id,
|
|
eventType: IIOS_EVENTS.messageSent,
|
|
cloudEvent: event as unknown as Prisma.InputJsonValue,
|
|
partitionKey: `${thread.scopeId}:${threadId}`,
|
|
},
|
|
});
|
|
|
|
// Bump unread for every participant except the sender.
|
|
const participants = await tx.iiosThreadParticipant.findMany({ where: { threadId } });
|
|
for (const p of participants) {
|
|
if (p.actorId === actor.id) continue;
|
|
await tx.iiosUnreadCounter.upsert({
|
|
where: { threadId_actorId: { threadId, actorId: p.actorId } },
|
|
create: { threadId, actorId: p.actorId, unreadCount: 1 },
|
|
update: { unreadCount: { increment: 1 } },
|
|
});
|
|
}
|
|
|
|
return tx.iiosInteraction.findUniqueOrThrow({
|
|
where: { id: created.id },
|
|
include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } },
|
|
});
|
|
});
|
|
|
|
return this.toDto(interaction, threadId);
|
|
} catch (err) {
|
|
if (err instanceof Prisma.PrismaClientKnownRequestError && err.code === 'P2002') {
|
|
const winner = await this.prisma.iiosInteraction.findUniqueOrThrow({
|
|
where: { scopeId_idempotencyKey: { scopeId: thread.scopeId, idempotencyKey } },
|
|
include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } },
|
|
});
|
|
return this.toDto(winner, threadId);
|
|
}
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
async markRead(
|
|
threadId: string,
|
|
principal: MessagePrincipal,
|
|
interactionId: string,
|
|
): 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.actors.resolveActor(thread.scopeId, principal);
|
|
|
|
const readEvent: CloudEvent = {
|
|
specversion: '1.0',
|
|
id: `evt_read_${interactionId}_${actor.id}`,
|
|
type: IIOS_EVENTS.messageRead,
|
|
source: `iios/message/${thread.scopeId}`,
|
|
subject: `interaction/${interactionId}`,
|
|
time: new Date().toISOString(),
|
|
datacontenttype: 'application/json',
|
|
insignia: { scopeSnapshotId: thread.scopeId, idempotencyKey: `read:${interactionId}:${actor.id}`, dataClass: 'internal' },
|
|
data: { threadId, actorId: actor.id, interactionId },
|
|
};
|
|
|
|
await this.prisma.$transaction([
|
|
this.prisma.iiosMessageReceipt.upsert({
|
|
where: { interactionId_actorId_receiptKind: { interactionId, actorId: actor.id, receiptKind: 'READ' } },
|
|
create: { interactionId, actorId: actor.id, receiptKind: 'READ' },
|
|
update: { occurredAt: new Date() },
|
|
}),
|
|
this.prisma.iiosUnreadCounter.upsert({
|
|
where: { threadId_actorId: { threadId, actorId: actor.id } },
|
|
create: { threadId, actorId: actor.id, unreadCount: 0, lastReadInteractionId: interactionId },
|
|
update: { unreadCount: 0, lastReadInteractionId: interactionId },
|
|
}),
|
|
this.prisma.iiosOutboxEvent.create({
|
|
data: {
|
|
aggregateType: 'interaction',
|
|
aggregateId: interactionId,
|
|
eventType: IIOS_EVENTS.messageRead,
|
|
cloudEvent: readEvent as unknown as Prisma.InputJsonValue,
|
|
partitionKey: `${thread.scopeId}:${threadId}`,
|
|
},
|
|
}),
|
|
]);
|
|
return { interactionId, actorId: actor.id };
|
|
}
|
|
|
|
async markDelivered(
|
|
threadId: string,
|
|
principal: MessagePrincipal,
|
|
interactionId: string,
|
|
): 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.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' },
|
|
update: { occurredAt: new Date() },
|
|
});
|
|
return { interactionId, actorId: actor.id };
|
|
}
|
|
|
|
async getMessageById(id: string, principal?: MessagePrincipal): Promise<MessageDto | null> {
|
|
const i = await this.prisma.iiosInteraction.findUnique({
|
|
where: { id },
|
|
include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } },
|
|
});
|
|
if (!i || !i.threadId) return null;
|
|
if (principal) await this.actors.assertOwns(principal, i.scopeId); // tenant fence (KG-02) when called on behalf of a caller
|
|
return this.toDto(i, i.threadId);
|
|
}
|
|
|
|
async history(threadId: string): Promise<MessageDto[]> {
|
|
const interactions = await this.prisma.iiosInteraction.findMany({
|
|
where: { threadId },
|
|
orderBy: { occurredAt: 'asc' },
|
|
include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } },
|
|
});
|
|
const annotations = await this.annotationsByInteraction(interactions.map((i) => i.id));
|
|
return interactions.map((i) => this.toDto(i, threadId, annotations.get(i.id) ?? []));
|
|
}
|
|
|
|
/**
|
|
* Toggle a generic annotation (actor attaches/removes an opaque label on an interaction).
|
|
* `annotationType`/`value` are app-supplied and never interpreted here (the chat app uses
|
|
* type "reaction" + an emoji value). Governed: only a thread participant may annotate.
|
|
* Returns the refreshed user list for that (type,value) so callers can broadcast it.
|
|
*/
|
|
async toggleAnnotation(
|
|
interactionId: string,
|
|
principal: MessagePrincipal,
|
|
annotationType: string,
|
|
value: string,
|
|
): Promise<{ interactionId: string; threadId: string; type: string; value: string; op: 'add' | 'remove'; users: string[] }> {
|
|
const target = await this.prisma.iiosInteraction.findUnique({
|
|
where: { id: interactionId },
|
|
select: { id: true, threadId: true, scopeId: true },
|
|
});
|
|
if (!target || !target.threadId) throw new NotFoundException('message not found');
|
|
const actor = await this.actors.resolveActor(target.scopeId, principal);
|
|
const isMember =
|
|
(await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId: target.threadId, actorId: actor.id } } })) !== null;
|
|
await decideOrThrow(this.ports, { action: 'iios.interaction.annotate', threadId: target.threadId, scopeId: target.scopeId, isMember });
|
|
|
|
const key = {
|
|
scopeId_targetInteractionId_actorId_annotationType_value: {
|
|
scopeId: target.scopeId,
|
|
targetInteractionId: interactionId,
|
|
actorId: actor.id,
|
|
annotationType,
|
|
value,
|
|
},
|
|
};
|
|
const existing = await this.prisma.iiosInteractionAnnotation.findUnique({ where: key });
|
|
let op: 'add' | 'remove';
|
|
if (existing) {
|
|
await this.prisma.iiosInteractionAnnotation.delete({ where: { id: existing.id } });
|
|
op = 'remove';
|
|
} else {
|
|
await this.prisma.iiosInteractionAnnotation.create({
|
|
data: { scopeId: target.scopeId, targetInteractionId: interactionId, actorId: actor.id, annotationType, value },
|
|
});
|
|
op = 'add';
|
|
}
|
|
|
|
const users =
|
|
(await this.annotationsByInteraction([interactionId])).get(interactionId)?.find((a) => a.type === annotationType && a.value === value)?.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 ────────────────────────────────────────────────────
|
|
|
|
/** Aggregate annotations for the given interactions into { type, value, users[] } groups. */
|
|
private async annotationsByInteraction(interactionIds: string[]): Promise<Map<string, AnnotationDto[]>> {
|
|
const map = new Map<string, AnnotationDto[]>();
|
|
if (interactionIds.length === 0) return map;
|
|
const rows = await this.prisma.iiosInteractionAnnotation.findMany({
|
|
where: { targetInteractionId: { in: interactionIds } },
|
|
include: { actor: { include: { sourceHandle: true } } },
|
|
orderBy: { createdAt: 'asc' },
|
|
});
|
|
for (const r of rows) {
|
|
const user = r.actor?.sourceHandle?.externalId ?? r.actor?.displayName ?? r.actorId;
|
|
const list = map.get(r.targetInteractionId) ?? [];
|
|
let entry = list.find((e) => e.type === r.annotationType && e.value === r.value);
|
|
if (!entry) {
|
|
entry = { type: r.annotationType, value: r.value, users: [] };
|
|
list.push(entry);
|
|
}
|
|
entry.users.push(user);
|
|
map.set(r.targetInteractionId, list);
|
|
}
|
|
for (const list of map.values()) for (const e of list) e.users.sort(); // deterministic order
|
|
return map;
|
|
}
|
|
|
|
private toDto(
|
|
interaction: {
|
|
id: string;
|
|
actorId: string | null;
|
|
actor?: { displayName: string | null; sourceHandle: { externalId: string } | null } | null;
|
|
parentInteractionId?: string | null;
|
|
traceId: string | null;
|
|
occurredAt: Date;
|
|
parts: Array<{ kind: string; bodyText: string | null; contentRef: string | null; mimeType?: string | null; sizeBytes?: bigint | null }>;
|
|
},
|
|
threadId: string,
|
|
annotations: AnnotationDto[] = [],
|
|
): MessageDto {
|
|
const text = interaction.parts.find((p) => p.kind === 'TEXT');
|
|
const file = interaction.parts.find((p) => p.contentRef);
|
|
const attachment: AttachmentDto | undefined = file?.contentRef
|
|
? {
|
|
contentRef: file.contentRef,
|
|
mimeType: file.mimeType ?? 'application/octet-stream',
|
|
sizeBytes: file.sizeBytes != null ? Number(file.sizeBytes) : 0,
|
|
kind: mediaKind(file.mimeType ?? ''),
|
|
}
|
|
: undefined;
|
|
return {
|
|
id: interaction.id,
|
|
threadId,
|
|
senderActorId: interaction.actorId ?? '',
|
|
senderId: interaction.actor?.sourceHandle?.externalId ?? '',
|
|
senderName: interaction.actor?.displayName ?? interaction.actor?.sourceHandle?.externalId ?? 'unknown',
|
|
content: text?.bodyText ?? '',
|
|
contentRef: file?.contentRef ?? undefined,
|
|
attachment,
|
|
parentInteractionId: interaction.parentInteractionId ?? undefined,
|
|
annotations,
|
|
traceId: interaction.traceId ?? '',
|
|
createdAt: interaction.occurredAt,
|
|
};
|
|
}
|
|
}
|