/**
 * Message persistence + read cursors.
 *
 * Durability rule: a message is written to Postgres *before* it is broadcast.
 * The socket layer only ever emits rows that already exist, so a client that
 * reconnects and asks for `seq > lastKnown` can never miss one.
 */
import { many, one, query } from '../../db/pool.js';
import { storage } from '../../storage/index.js';
import { env } from '../../config/env.js';
import { badRequest, notFound } from '../../utils/errors.js';
import type { AttachmentDTO, MessageDTO, SenderType } from '../types.js';

const MAX_BODY = 8000;

export interface CreateMessageInput {
  conversationId: string;
  senderType: SenderType;
  senderId?: string | null;
  body?: string;
  type?: 'text' | 'file' | 'system';
  attachmentId?: string | null;
  clientMsgId?: string | null;
}

const MESSAGE_SELECT = `
  SELECT m.id, m.seq, m.conversation_id, m.sender_type, m.sender_id, m.type, m.body,
         m.client_msg_id, m.created_at,
         COALESCE(a.name, cu.name) AS sender_name,
         at.id AS att_id, at.original_name, at.mime_type, at.size_bytes, at.storage_key
    FROM messages m
    LEFT JOIN agents a     ON a.id = m.sender_id AND m.sender_type = 'agent'
    LEFT JOIN customers cu ON cu.id = m.sender_id AND m.sender_type = 'customer'
    LEFT JOIN attachments at ON at.id = m.attachment_id
`;

async function toMessageDTO(row: any): Promise<MessageDTO> {
  let attachment: AttachmentDTO | null = null;
  if (row.att_id) {
    const mime = row.mime_type as string;
    attachment = {
      id: row.att_id,
      name: row.original_name,
      mimeType: mime,
      size: Number(row.size_bytes),
      // Signed and short-lived: the URL in a persisted message is regenerated
      // on every read rather than stored.
      url: await storage
        .downloadUrl(row.storage_key, env.FILE_URL_TTL, row.original_name)
        .catch(() => null),
      isImage: mime.startsWith('image/'),
    };
  }
  return {
    id: row.id,
    seq: Number(row.seq),
    conversationId: row.conversation_id,
    senderType: row.sender_type,
    senderId: row.sender_id ?? null,
    senderName: row.sender_name ?? (row.sender_type === 'system' ? 'System' : null),
    type: row.type,
    body: row.body ?? '',
    attachment,
    clientMsgId: row.client_msg_id ?? null,
    createdAt: new Date(row.created_at).toISOString(),
  };
}

/**
 * Insert a message. Idempotent on (conversationId, clientMsgId): a client that
 * retries after a dropped ack gets the original row back, not a duplicate.
 */
export async function createMessage(input: CreateMessageInput): Promise<MessageDTO> {
  const body = (input.body ?? '').slice(0, MAX_BODY);
  const type = input.type ?? (input.attachmentId ? 'file' : 'text');
  if (type === 'text' && !body.trim()) throw badRequest('Message body cannot be empty');

  const inserted = await one(
    `INSERT INTO messages (conversation_id, sender_type, sender_id, type, body, attachment_id, client_msg_id)
     VALUES ($1,$2,$3,$4,$5,$6,$7)
     ON CONFLICT (conversation_id, client_msg_id) WHERE client_msg_id IS NOT NULL
     DO UPDATE SET body = messages.body    -- no-op, forces RETURNING to fire
     RETURNING id`,
    [
      input.conversationId,
      input.senderType,
      input.senderId ?? null,
      type,
      body,
      input.attachmentId ?? null,
      input.clientMsgId ?? null,
    ],
  );
  if (!inserted) throw badRequest('Failed to persist message');

  await query(`UPDATE conversations SET last_message_at = now() WHERE id = $1`, [input.conversationId]);

  const row = await one(`${MESSAGE_SELECT} WHERE m.id = $1`, [inserted.id]);
  return toMessageDTO(row);
}

/** Newest page first is wrong for a transcript — always return ascending by seq. */
export async function listMessages(
  conversationId: string,
  opts: { sinceSeq?: number; beforeSeq?: number; limit?: number } = {},
): Promise<MessageDTO[]> {
  const limit = Math.min(Math.max(opts.limit ?? 100, 1), 500);

  if (opts.beforeSeq) {
    // Backwards pagination ("load older") — fetch descending then flip.
    const rows = await many(
      `${MESSAGE_SELECT} WHERE m.conversation_id = $1 AND m.seq < $2 ORDER BY m.seq DESC LIMIT $3`,
      [conversationId, opts.beforeSeq, limit],
    );
    return Promise.all(rows.reverse().map(toMessageDTO));
  }

  const rows = await many(
    `${MESSAGE_SELECT} WHERE m.conversation_id = $1 AND m.seq > $2 ORDER BY m.seq ASC LIMIT $3`,
    [conversationId, opts.sinceSeq ?? 0, limit],
  );
  return Promise.all(rows.map(toMessageDTO));
}

/** The last `limit` messages — what a freshly opened chat window loads. */
export async function listRecentMessages(conversationId: string, limit = 100): Promise<MessageDTO[]> {
  const rows = await many(
    `${MESSAGE_SELECT} WHERE m.conversation_id = $1 ORDER BY m.seq DESC LIMIT $2`,
    [conversationId, limit],
  );
  return Promise.all(rows.reverse().map(toMessageDTO));
}

export async function getMessage(id: string): Promise<MessageDTO> {
  const row = await one(`${MESSAGE_SELECT} WHERE m.id = $1`, [id]);
  if (!row) throw notFound('Message not found');
  return toMessageDTO(row);
}

/**
 * Advance a read cursor. GREATEST() keeps it monotonic so an out-of-order
 * receipt from a slow tab can never un-read messages.
 */
export async function markRead(
  conversationId: string,
  reader: 'customer' | 'agent',
  seq: number,
): Promise<number> {
  const column = reader === 'agent' ? 'agent_last_read_seq' : 'customer_last_read_seq';
  const row = await one(
    `UPDATE conversations SET ${column} = GREATEST(${column}, $2)
      WHERE id = $1 RETURNING ${column} AS seq`,
    [conversationId, seq],
  );
  return Number(row?.seq ?? 0);
}

export async function maxSeq(conversationId: string): Promise<number> {
  const row = await one(`SELECT COALESCE(MAX(seq), 0) AS seq FROM messages WHERE conversation_id = $1`, [
    conversationId,
  ]);
  return Number(row?.seq ?? 0);
}

/** System messages narrate lifecycle events inside the transcript itself. */
export function systemMessage(conversationId: string, body: string): Promise<MessageDTO> {
  return createMessage({ conversationId, senderType: 'system', type: 'system', body });
}