iios/packages/iios-service/src/inbox/inbox.projector.ts

import { Injectable, OnModuleInit } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts';
import { PrismaService } from '../prisma/prisma.service';
import { OutboxBus } from '../outbox/outbox.bus';
import { DlqService } from '../outbox/dlq.service';
import { ProjectionCursorService } from '../projection/projection-cursor.service';

interface MessageSentData {
  interactionId: string;
  threadId: string;
  senderActorId: string;
}
interface MessageReadData {
  threadId: string;
  actorId: string;
  interactionId: string;
}

/**
 * Projects message events into Inbox work-items (P3). Inbox reacts to events —
 * it never imports the message layer. Deterministic + idempotent (collapse one
 * OPEN NEEDS_REPLY per owner+thread; dedupe each event via the processed-event
 * ledger). Reading a thread resolves the owner's item to DONE.
 */
@Injectable()
export class InboxProjector implements OnModuleInit {
  private readonly consumer = 'inbox-projector';

  constructor(
    private readonly prisma: PrismaService,
    private readonly bus: OutboxBus,
    private readonly dlq: DlqService,
    private readonly cursor: ProjectionCursorService,
  ) {}

  onModuleInit(): void {
    this.dlq.registerHandler(this.consumer, (e) => this.onMessageSent(e));
    this.bus.on(IIOS_EVENTS.messageSent, (p) => void this.onMessageSent(p as CloudEvent).catch((err) => this.dlq.onConsumerFailure(this.consumer, p as CloudEvent, err)));
    this.bus.on(IIOS_EVENTS.messageRead, (p) => void this.onMessageRead(p as CloudEvent).catch((err) => this.dlq.onConsumerFailure(this.consumer, p as CloudEvent, err)));
  }

  async onMessageSent(event: CloudEvent): Promise<void> {
    if (!(await this.claim(event.id))) return; // duplicate → no apply, no cursor advance
    await this.applyMessageSent(event);
    await this.cursor.advance(this.consumer, event);
  }

  private async applyMessageSent(event: CloudEvent): Promise<void> {
    const data = event.data as MessageSentData;
    const traceId = event.insignia?.correlationId;
    const thread = await this.prisma.iiosThread.findUnique({ where: { id: data.threadId } });
    if (!thread) return;

    const participants = await this.prisma.iiosThreadParticipant.findMany({ where: { threadId: data.threadId } });
    for (const p of participants) {
      if (p.actorId === data.senderActorId) continue;
      await this.upsertNeedsReply(thread.scopeId, p.actorId, data.threadId, data.interactionId, traceId, thread.subject);
    }
  }

  async onMessageRead(event: CloudEvent): Promise<void> {
    if (!(await this.claim(event.id))) return; // duplicate → no apply, no cursor advance
    await this.applyMessageRead(event);
    await this.cursor.advance(this.consumer, event);
  }

  private async applyMessageRead(event: CloudEvent): Promise<void> {
    const data = event.data as MessageReadData;
    const item = await this.prisma.iiosInboxItem.findFirst({
      where: {
        threadId: data.threadId,
        ownerActorId: data.actorId,
        kind: 'NEEDS_REPLY',
        state: { in: ['OPEN', 'SNOOZED'] },
      },
    });
    if (!item) return;
    await this.prisma.$transaction([
      this.prisma.iiosInboxItem.update({ where: { id: item.id }, data: { state: 'DONE' } }),
      this.prisma.iiosInboxItemStateHistory.create({
        data: { inboxItemId: item.id, fromState: item.state, toState: 'DONE', reasonCode: 'message_read' },
      }),
    ]);
  }

  private async upsertNeedsReply(
    scopeId: string,
    ownerActorId: string,
    threadId: string,
    sourceInteractionId: string,
    traceId: string | undefined,
    subject: string | null,
  ): Promise<void> {
    const existing = await this.prisma.iiosInboxItem.findFirst({
      where: { ownerActorId, threadId, kind: 'NEEDS_REPLY', state: { in: ['OPEN', 'SNOOZED'] } },
    });
    if (existing) {
      await this.prisma.iiosInboxItem.update({
        where: { id: existing.id },
        data: { sourceInteractionId, traceId: traceId ?? existing.traceId, state: 'OPEN' },
      });
      return;
    }

    const item = await this.prisma.iiosInboxItem.create({
      data: {
        scopeId,
        ownerActorId,
        kind: 'NEEDS_REPLY',
        title: subject ?? 'New messages',
        threadId,
        sourceInteractionId,
        traceId,
      },
    });
    await this.prisma.iiosInboxItemStateHistory.create({
      data: { inboxItemId: item.id, toState: 'OPEN', reasonCode: 'message_sent' },
    });

    const created: CloudEvent = {
      specversion: '1.0',
      id: `evt_${item.id}`,
      type: IIOS_EVENTS.inboxItemCreated,
      source: `iios/inbox/${scopeId}`,
      subject: `inbox_item/${item.id}`,
      time: new Date().toISOString(),
      datacontenttype: 'application/json',
      insignia: { scopeSnapshotId: scopeId, correlationId: traceId, idempotencyKey: `inbox:${item.id}`, dataClass: 'internal' },
      data: { inboxItemId: item.id, ownerActorId, threadId, kind: 'NEEDS_REPLY' },
    };
    await this.prisma.iiosOutboxEvent.create({
      data: {
        aggregateType: 'inbox_item',
        aggregateId: item.id,
        eventType: IIOS_EVENTS.inboxItemCreated,
        cloudEvent: created as unknown as Prisma.InputJsonValue,
        partitionKey: `${scopeId}:${threadId}`,
      },
    });
  }

  /** Idempotent claim: false if this event id was already processed. */
  private async claim(eventId: string): Promise<boolean> {
    const res = await this.prisma.iiosProcessedEvent.createMany({
      data: [{ consumerName: this.consumer, eventId }],
      skipDuplicates: true,
    });
    return res.count > 0;
  }
}