import { randomUUID } from 'node:crypto';
import { BadRequestException, 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 {
  /** Ceiling on one bulk participant import — bounds the work per request and the blast radius of a
   *  mistaken import. Raise here if a larger roster copy is ever needed. */
  static readonly MAX_BULK_PARTICIPANTS = 200;

  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 };
  }

  /**
   * Bulk sibling of {@link addParticipant}: add many users in ONE governed operation. The app uses
   * this to import a roster (e.g. "add everyone from that channel"); the kernel stays generic — it
   * receives an explicit list of userIds and never learns where the list came from.
   *
   * Deliberately NOT atomic: a single unresolvable user must not sink the whole import, so each is
   * attempted independently and the outcome is reported per user. Already-members are `skipped`,
   * which makes a re-run a no-op.
   */
  async addParticipants(
    threadId: string,
    principal: MessagePrincipal,
    targetUserIds: string[],
    role = 'MEMBER',
  ): Promise<{ threadId: string; added: string[]; skipped: string[]; failed: string[]; participantCount: number }> {
    const unique = [...new Set(targetUserIds.map((u) => u.trim()).filter(Boolean))];
    if (unique.length === 0) throw new BadRequestException('at least one userId is required');
    if (unique.length > MessageService.MAX_BULK_PARTICIPANTS) {
      throw new BadRequestException(`at most ${MessageService.MAX_BULK_PARTICIPANTS} participants can be added at once`);
    }

    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;

    // ONE decision for the whole batch — `targetCount` lets policy reason about the resulting size
    // (e.g. the dm two-person cap) instead of being asked the same question N times.
    await decideOrThrow(this.ports, {
      action: 'iios.thread.participant.add',
      threadId,
      scopeId: thread.scopeId,
      membership,
      participantCount,
      callerRole: callerP?.participantRole,
      targetCount: unique.length,
      role,
    });

    const scope = await this.actors.resolveScope(principal);
    const added: string[] = [];
    const skipped: string[] = [];
    const failed: string[] = [];
    for (const targetUserId of unique) {
      try {
        const target = await this.actors.resolveActor(scope.id, {
          userId: targetUserId,
          appId: principal.appId,
          orgId: principal.orgId,
          tenantId: principal.tenantId,
          displayName: targetUserId,
        });
        const existing = await this.prisma.iiosThreadParticipant.findUnique({
          where: { threadId_actorId: { threadId, actorId: target.id } },
        });
        if (existing) {
          skipped.push(targetUserId);
          continue;
        }
        await this.actors.ensureParticipant(threadId, target.id, role);
        added.push(targetUserId);
      } catch {
        failed.push(targetUserId);
      }
    }
    return { threadId, added, skipped, failed, participantCount: participantCount + added.length };
  }

  /**
   * 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');
    // Roster reads are membership-governed (a private channel's members must not be enumerable by
    // a non-member), so the decision carries the caller's role + the thread's opaque attributes.
    const caller = await this.actors.resolveActor(thread.scopeId, principal);
    const callerP = await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: caller.id } } });
    const meta = thread.metadata as { membership?: string; visibility?: string } | null;
    await decideOrThrow(this.ports, {
      action: 'iios.thread.participant.list',
      threadId,
      scopeId: thread.scopeId,
      membership: meta?.membership,
      visibility: meta?.visibility,
      callerRole: callerP?.participantRole,
    });
    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,
    };
  }
}
