apps/api/src/conversations/conversations.service.ts

/**
 * ConversationsService — core ingestion service for inbound messages from all
 * channels. Implements the INBOUND ONLY path for Day 2 (Track 2a).
 *
 * Key invariants:
 * - ONE transaction with advisory lock per conversation_id (§30.16 + §3.5).
 * - Media download/storage runs AFTER the transaction (network I/O outside txn).
 * - Does NOT emit Inngest events — the caller (webhook controller) does that.
 * - No fake sends: Day 2 is ingestion only (CLAUDE.md #2).
 */
import { Inject, Injectable, Logger } from "@nestjs/common";
import { and, eq, sql } from "drizzle-orm";
import { createHash } from "node:crypto";
import {
  type Database,
  users,
  userIdentities,
  conversations,
  messages,
  missions,
  missionEvents,
  missionStatusSnapshots,
  files,
  evidenceItems,
} from "@courtmitra/db";
import type { NormalizedInboundMessage } from "@courtmitra/contracts";
import type { StoragePort, TranslationPort } from "@courtmitra/ports";
import { DB } from "../db/db.module.js";
import { downloadTelegramFile, downloadWhatsAppFile } from "@courtmitra/adapters";

export const STORAGE = Symbol("STORAGE");
export const TRANSLATION = Symbol("TRANSLATION");

export interface IngestResult {
  missionId: string;
  conversationId: string;
  messageId: string;
  createdMission: boolean;
  evidenceItemIds: string[];
}

/** Derive evidence type from media kind. */
function evidenceTypeFromKind(kind: NormalizedInboundMessage["media"][number]["kind"]): string {
  switch (kind) {
    case "photo":
      return "screenshot";
    case "document":
      return "pdf";
    case "audio":
      return "audio";
    case "video":
      return "video";
    default:
      return "other";
  }
}

/** Initial mission status snapshot (arch §30.6). */
function intakeSnapshot(missionId: string) {
  return {
    missionId,
    state: "intake_in_progress" as const,
    substate: null,
    headline: "We received your message — processing your grievance.",
    blockedReason: null,
    nextUserAction: "We'll ask follow-up questions if we need more info.",
    nextSystemAction: "Run intake to structure your grievance.",
    primaryButton: "View status",
    danger: false,
    confidence: null,
  };
}

@Injectable()
export class ConversationsService {
  private readonly logger = new Logger(ConversationsService.name);

  constructor(
    @Inject(DB) private readonly db: Database,
    @Inject(STORAGE) private readonly storage: StoragePort,
    @Inject(TRANSLATION) private readonly translation: TranslationPort,
  ) {}

  /**
   * Ingest an inbound message end-to-end:
   *   1. Advisory lock on conversationId (hashtext of externalThreadId+channel).
   *   2. get-or-create user + user_identity.
   *   3. get-or-create conversation.
   *   4. get-or-create mission (sets active_mission_id when creating).
   *   5. insert message.
   *   6. append message.received event.
   *   All in ONE transaction.
   *
   * Media is handled AFTER the transaction: download → storage.put → files →
   * evidence_items rows.
   */
  async ingestInbound(
    msg: NormalizedInboundMessage,
    opts: { issueDomainGuess?: string } = {},
  ): Promise<IngestResult> {
    // -----------------------------------------------------------------------
    // Detect language BEFORE the transaction (network I/O shouldn't hold lock)
    // -----------------------------------------------------------------------
    let detectedLanguage = "en";
    if (msg.text && msg.text.trim().length > 0) {
      try {
        const detection = await this.translation.detectLanguage(msg.text);
        detectedLanguage = detection.language;
      } catch {
        // If detection fails, default to English
      }
    }

    // -----------------------------------------------------------------------
    // Phase 1: transactional ingestion with advisory lock
    // -----------------------------------------------------------------------
    const { missionId, conversationId, messageId, createdMission } = await this.db.transaction(
      async (tx) => {
        // Advisory lock keyed by (channel + externalThreadId) to prevent
        // duplicate mission creation from concurrent messages on the same chat.
        await tx.execute(
          sql`SELECT pg_advisory_xact_lock(hashtext(${msg.channel + ":" + msg.externalThreadId}))`,
        );

        // 1. get-or-create user
        const existingIdentity = await tx
          .select({ userId: userIdentities.userId })
          .from(userIdentities)
          .where(
            and(
              eq(userIdentities.provider, msg.channel),
              eq(userIdentities.providerSubject, msg.fromSubject),
            ),
          )
          .limit(1);

        let userId: string;
        if (existingIdentity[0]) {
          userId = existingIdentity[0].userId;
        } else {
          const [newUser] = await tx
            .insert(users)
            .values({ displayName: null, preferredLanguage: detectedLanguage })
            .returning();
          userId = newUser!.id;
          await tx.insert(userIdentities).values({
            userId,
            provider: msg.channel,
            providerSubject: msg.fromSubject,
            isVerified: false,
            metadata: null,
          });
        }

        // 2. get-or-create conversation
        const existingConv = await tx
          .select()
          .from(conversations)
          .where(
            and(
              eq(conversations.channel, msg.channel),
              eq(conversations.externalThreadId, msg.externalThreadId),
            ),
          )
          .limit(1);

        let conversationId: string;
        let activeMissionId: string | null = null;

        if (existingConv[0]) {
          conversationId = existingConv[0].id;
          activeMissionId = existingConv[0].activeMissionId ?? null;
        } else {
          // Fix 6: use onConflictDoNothing + re-select so a lost race returns
          // the existing row instead of a duplicate or error.
          const [newConv] = await tx
            .insert(conversations)
            .values({
              userId,
              channel: msg.channel,
              externalThreadId: msg.externalThreadId,
              activeMissionId: null,
              lastMessageAt: new Date(),
            })
            .onConflictDoNothing()
            .returning();

          if (newConv) {
            conversationId = newConv.id;
          } else {
            // Lost the race — re-select the winner.
            const [raceWinner] = await tx
              .select()
              .from(conversations)
              .where(
                and(
                  eq(conversations.channel, msg.channel),
                  eq(conversations.externalThreadId, msg.externalThreadId),
                ),
              )
              .limit(1);
            if (!raceWinner) throw new Error(`Conversation race-condition lookup failed for ${msg.channel}:${msg.externalThreadId}`);
            conversationId = raceWinner.id;
            activeMissionId = raceWinner.activeMissionId ?? null;
          }
        }

        // 3. get-or-create mission
        let createdMission = false;
        let missionId: string;

        if (activeMissionId) {
          const [activeMission] = await tx
            .select({ state: missions.state })
            .from(missions)
            .where(eq(missions.id, activeMissionId))
            .limit(1);

          const isClosed = !activeMission || [
            "resolved",
            "closed_unresolved",
            "resolution_claimed",
            "failed"
          ].includes(activeMission.state);

          const isStartCmd = msg.text?.trim().startsWith("/start") || msg.text?.trim().startsWith("/new");

          if (isClosed || isStartCmd) {
            activeMissionId = null;
          }
        }

        if (activeMissionId) {
          missionId = activeMissionId;
        } else {
          const isCmd = msg.text?.trim().startsWith("/");
          const titleText = isCmd ? "New grievance" : (msg.text?.slice(0, 120) ?? "New grievance");
          const [newMission] = await tx
            .insert(missions)
            .values({
              userId,
              title: titleText,
              issueDomain: opts.issueDomainGuess ?? "consumer",
              state: "intake_in_progress",
              safetyLevel: "normal",
            })
            .returning();
          missionId = newMission!.id;
          createdMission = true;

          // Set active_mission_id on conversation
          await tx
            .update(conversations)
            .set({ activeMissionId: missionId, lastMessageAt: new Date() })
            .where(eq(conversations.id, conversationId));

          // Append mission.created event
          await tx.insert(missionEvents).values({
            missionId,
            eventType: "mission.created",
            actorType: "system",
            actorId: null,
            payload: {
              source: msg.channel,
              externalThreadId: msg.externalThreadId,
              issueDomainGuess: opts.issueDomainGuess ?? "consumer",
            },
          });

          // Initial status snapshot
          await tx.insert(missionStatusSnapshots).values(intakeSnapshot(missionId));
        }

        // 4. Insert message
        const idempotencyKey = `${msg.channel}:${msg.providerMessageId}`;
        const mediaRefs = msg.media.map((m) => ({
          kind: m.kind,
          providerFileId: m.providerFileId,
          mimeType: m.mimeType,
          filename: m.filename,
        }));

        const [newMessage] = await tx
          .insert(messages)
          .values({
            conversationId,
            missionId,
            direction: "inbound",
            channel: msg.channel,
            channelMessageId: msg.providerMessageId,
            idempotencyKey,
            body: msg.text,
            mediaRefs: mediaRefs.length > 0 ? mediaRefs : null,
            rawPayload: msg.rawPayloadRef as Record<string, unknown>,
          })
          .onConflictDoNothing()
          .returning();

        const messageId = newMessage?.id;
        if (!messageId) {
          // Already inserted via idempotency — look up the existing message
          const [existing] = await tx
            .select({ id: messages.id })
            .from(messages)
            .where(eq(messages.idempotencyKey, idempotencyKey))
            .limit(1);
          if (!existing) throw new Error(`Message idempotency lookup failed for key ${idempotencyKey}`);
          return { missionId, conversationId, messageId: existing.id, createdMission };
        }

        // 5. Append message.received event
        await tx.insert(missionEvents).values({
          missionId,
          eventType: "message.received",
          actorType: "user",
          actorId: userId,
          payload: { messageId, channel: msg.channel, hasText: !!msg.text, mediaCount: msg.media.length },
        });

        // Update conversation lastMessageAt
        await tx
          .update(conversations)
          .set({ lastMessageAt: new Date() })
          .where(eq(conversations.id, conversationId));

        return { missionId, conversationId, messageId, createdMission };
      },
    );

    // -----------------------------------------------------------------------
    // Phase 2: media handling — outside the transaction (network I/O)
    // -----------------------------------------------------------------------
    const evidenceItemIds: string[] = [];

    for (const media of msg.media) {
      try {
        let bytes: Uint8Array;
        let mimeType: string | null = media.mimeType;
        let filename: string | null = media.filename;

        if (msg.channel === "telegram") {
          const downloaded = await downloadTelegramFile(media.providerFileId);
          bytes = downloaded.bytes;
          if (downloaded.mimeType && downloaded.mimeType !== "application/octet-stream") {
            mimeType = downloaded.mimeType;
          } else if (!mimeType) {
            mimeType = downloaded.mimeType;
          }
          filename = downloaded.filename ?? filename;
        } else if (msg.channel === "whatsapp") {
          const downloaded = await downloadWhatsAppFile(media.providerFileId);
          bytes = downloaded.bytes;
          mimeType = downloaded.mimeType ?? mimeType;
        } else {
          // Unknown channel: skip media download
          this.logger.warn(`Media download not supported for channel: ${msg.channel}`);
          continue;
        }

        const sha256 = createHash("sha256").update(bytes).digest("hex");
        const ext = filename?.split(".").pop() ?? "bin";
        const storageKey = `evidence/${missionId}/${media.providerFileId}.${ext}`;

        await this.storage.put({ key: storageKey, bytes, mimeType: mimeType ?? undefined });

        const [fileRow] = await this.db
          .insert(files)
          .values({
            missionId,
            uploaderUserId: null,
            storageKey,
            originalFilename: filename,
            mimeType,
            sha256,
            sizeBytes: bytes.length,
            processingStatus: "pending",
            // Fix 11: honest — no virus scanner in Phase 1.
            virusScanStatus: "skipped",
          })
          .returning();

        const [evItem] = await this.db
          .insert(evidenceItems)
          .values({
            missionId,
            fileId: fileRow!.id,
            uploaderUserId: null,
            type: evidenceTypeFromKind(media.kind),
            captureDate: null,
            sourceChannel: msg.channel,
            piiLevel: "unknown",
          })
          .returning();

        if (evItem) evidenceItemIds.push(evItem.id);
      } catch (err) {
        this.logger.error(`Failed to process media ${media.providerFileId}: ${String(err)}`);
      }
    }

    return { missionId, conversationId, messageId, createdMission, evidenceItemIds };
  }
}