feat(iios): policy-enforced membership + threaded replies + dev IdP (generic)
Enriches the platform PLANE, not the kernel: - DevOpaPort: the dev OPA stub now evaluates a small policy table (behind the same opa.decide) — DM capped at 2, add-participant requires member/group-admin, governed self-join for membership threads. Real OPA swaps in unchanged. - Dev IdP login (POST /v1/dev/login) issues the same JWT claims a real IdP would. - PolicyDeniedFilter maps fail-closed denials to HTTP 403. Generic kernel additions (no chat vocabulary — 'dm'/'group' live only as OPA policy + an opaque thread attribute): - MessageService.addParticipant (governed membership by userId), governed openThread self-join (scoped to threads with a membership attribute), parentInteractionId on send (reply link), and a generic listThreads. - REST: GET /v1/threads, POST /v1/threads, POST /v1/threads/:id/participants; socket add_participant + membership/parentInteractionId. ensureParticipant gains a role. Tests: dev-opa.port.spec + message.spec (DM cap / group admin / governed join / listThreads / reply). smoke-membership.mjs; realtime smokes updated for governed join. 175 unit tests + all smokes green; kernel free of dm/group literals. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -15,10 +15,21 @@ export interface MessageDto {
|
||||
senderActorId: string;
|
||||
content: string;
|
||||
contentRef?: string;
|
||||
parentInteractionId?: string;
|
||||
traceId: string;
|
||||
createdAt: Date;
|
||||
}
|
||||
|
||||
export interface ThreadSummary {
|
||||
threadId: string;
|
||||
subject: string | null;
|
||||
membership?: string;
|
||||
participantCount: number;
|
||||
unread: number;
|
||||
lastMessage?: string;
|
||||
lastAt?: Date;
|
||||
}
|
||||
|
||||
export interface OpenThreadResult {
|
||||
threadId: string;
|
||||
status: string;
|
||||
@@ -39,16 +50,28 @@ export class MessageService {
|
||||
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> {
|
||||
/**
|
||||
* 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 }): 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` are generic, app-supplied thread attributes — the kernel
|
||||
// stores/echoes them but never branches on their chat meaning (that lives in policy + app).
|
||||
const thread = await this.prisma.iiosThread.create({
|
||||
data: { scopeId: scope.id, createdByActorId: actor.id },
|
||||
data: {
|
||||
scopeId: scope.id,
|
||||
createdByActorId: actor.id,
|
||||
metadata: opts?.membership ? ({ membership: opts.membership } as Prisma.InputJsonValue) : undefined,
|
||||
},
|
||||
});
|
||||
await this.actors.ensureParticipant(thread.id, actor.id);
|
||||
await this.actors.ensureParticipant(thread.id, actor.id, opts?.creatorRole ?? 'MEMBER');
|
||||
return { threadId: thread.id, status: thread.status, history: [] };
|
||||
}
|
||||
|
||||
@@ -56,16 +79,96 @@ export class MessageService {
|
||||
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);
|
||||
// support/inbox/generic threads (no membership attr) keep open-join, unchanged.
|
||||
const membership = (thread.metadata as { membership?: string } | null)?.membership;
|
||||
await decideOrThrow(this.ports, { action: 'iios.thread.join', threadId, scopeId: thread.scopeId, membership, 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 };
|
||||
}
|
||||
|
||||
/** Generic "my threads": every thread the caller participates in, with last message + unread. */
|
||||
async listThreads(principal: MessagePrincipal): 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 } });
|
||||
const threadIds = memberships.map((m) => m.threadId);
|
||||
if (threadIds.length === 0) return [];
|
||||
|
||||
const [threads, unreads] = await Promise.all([
|
||||
this.prisma.iiosThread.findMany({ where: { id: { in: threadIds } }, include: { _count: { select: { participants: true } } } }),
|
||||
this.prisma.iiosUnreadCounter.findMany({ where: { threadId: { in: threadIds }, actorId: actor.id } }),
|
||||
]);
|
||||
const unreadBy = new Map(unreads.map((u) => [u.threadId, u.unreadCount]));
|
||||
|
||||
const summaries = await Promise.all(
|
||||
threads.map(async (t) => {
|
||||
const last = await this.prisma.iiosInteraction.findFirst({
|
||||
where: { threadId: t.id },
|
||||
orderBy: { occurredAt: 'desc' },
|
||||
include: { parts: { where: { kind: 'TEXT' }, take: 1 } },
|
||||
});
|
||||
return {
|
||||
threadId: t.id,
|
||||
subject: t.subject,
|
||||
membership: (t.metadata as { membership?: string } | null)?.membership,
|
||||
participantCount: t._count.participants,
|
||||
unread: unreadBy.get(t.id) ?? 0,
|
||||
lastMessage: last?.parts[0]?.bodyText ?? 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 },
|
||||
idempotencyKey: string,
|
||||
traceId: string = randomUUID(),
|
||||
parentInteractionId?: string,
|
||||
): Promise<MessageDto> {
|
||||
const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
|
||||
if (!thread) throw new NotFoundException('thread not found');
|
||||
@@ -75,6 +178,11 @@ export class MessageService {
|
||||
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 } },
|
||||
@@ -93,6 +201,7 @@ export class MessageService {
|
||||
idempotencyKey,
|
||||
status: 'NORMALIZED',
|
||||
traceId,
|
||||
parentInteractionId: parentRef,
|
||||
},
|
||||
});
|
||||
|
||||
@@ -239,7 +348,7 @@ export class MessageService {
|
||||
// ─── helpers ────────────────────────────────────────────────────
|
||||
|
||||
private toDto(
|
||||
interaction: { id: string; actorId: string | null; traceId: string | null; occurredAt: Date; parts: Array<{ kind: string; bodyText: string | null; contentRef: string | null }> },
|
||||
interaction: { id: string; actorId: string | null; parentInteractionId?: string | null; traceId: string | null; occurredAt: Date; parts: Array<{ kind: string; bodyText: string | null; contentRef: string | null }> },
|
||||
threadId: string,
|
||||
): MessageDto {
|
||||
const text = interaction.parts.find((p) => p.kind === 'TEXT');
|
||||
@@ -250,6 +359,7 @@ export class MessageService {
|
||||
senderActorId: interaction.actorId ?? '',
|
||||
content: text?.bodyText ?? '',
|
||||
contentRef: file?.contentRef ?? undefined,
|
||||
parentInteractionId: interaction.parentInteractionId ?? undefined,
|
||||
traceId: interaction.traceId ?? '',
|
||||
createdAt: interaction.occurredAt,
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user