import "server-only";

import { createHash } from "crypto";
import type { ResultSetHeader, RowDataPacket } from "mysql2/promise";
import { ensureOfficialMailSchema, getDbPool } from "@/lib/db";
import { enqueueMailboxAttachmentStorageDeletesByMessageIds } from "@/lib/mailbox-file-drawer";
import {
  extractAttachmentPayloadsFromRawSource,
  extractAttachmentsFromRawSource,
  type RemoteMailboxAttachmentMeta,
  type RemoteMailboxAttachmentPayload,
} from "@/lib/mailbox-remote";
import {
  readObjectFromObjectStorage,
  uploadPrivateBufferToObjectStorage,
} from "@/lib/object-storage";
import {
  activateMailboxStorageObject,
  reserveMailboxStorageObject,
  scheduleMailboxStorageObjectDeletion,
} from "@/lib/mailbox-storage-registry";

type AttachmentCacheRow = RowDataPacket & {
  attachment_index: number;
  content_disposition: "attachment" | "inline";
  content_id: string | null;
  content_sha256: string | null;
  mime_type: string;
  original_name: string;
  size_bytes: number;
  storage_key: string | null;
  storage_status: "metadata" | "cached" | "failed";
};

type MessageAttachmentStateRow = RowDataPacket & {
  attachments_indexed_at: Date | null;
  raw_source: string | null;
};

type AttachmentOwnerRow = RowDataPacket & {
  mailbox_email: string;
  mailbox_id: number;
  owner_email: string;
  owner_user_id: number;
};

type AttachmentReservationRow = RowDataPacket & {
  id: number;
  storage_key: string | null;
  storage_status: "metadata" | "cached" | "failed";
};

const attachmentCacheQueues = new Map<
  number,
  Promise<RemoteMailboxAttachmentPayload[]>
>();
const attachmentMetadataQueues = new Map<
  number,
  Promise<RemoteMailboxAttachmentMeta[]>
>();

function extensionFromFilename(filename: string) {
  return /\.([a-z0-9]{1,12})$/i.exec(filename.trim())?.[1]?.toLowerCase() ?? "";
}

function isPreviewableAttachment(contentType: string, filename: string) {
  const normalizedContentType = contentType.trim().toLowerCase();
  const extension = extensionFromFilename(filename);

  return (
    normalizedContentType.startsWith("image/") ||
    normalizedContentType.startsWith("text/") ||
    normalizedContentType === "application/pdf" ||
    ["csv", "gif", "jpeg", "jpg", "json", "md", "pdf", "png", "svg", "txt", "webp"].includes(
      extension,
    )
  );
}

function rowToMetadata(row: AttachmentCacheRow): RemoteMailboxAttachmentMeta {
  return {
    contentDisposition: row.content_disposition,
    contentId: row.content_id,
    contentType: row.mime_type,
    extension: extensionFromFilename(row.original_name),
    filename: row.original_name,
    index: Number(row.attachment_index),
    // MIME disposition alone is unreliable. The message body decides whether
    // this attachment is actually embedded before it is returned to the UI.
    isInline: false,
    isPreviewable: isPreviewableAttachment(row.mime_type, row.original_name),
    sizeBytes: Number(row.size_bytes),
  };
}

async function getAttachmentRows(messageId: number) {
  const [rows] = await getDbPool().query<AttachmentCacheRow[]>(
    `
      SELECT
        attachment_index,
        content_disposition,
        content_id,
        original_name,
        mime_type,
        size_bytes,
        storage_status,
        storage_key,
        content_sha256
      FROM mailbox_message_attachments
      WHERE mailbox_message_id = ?
      ORDER BY attachment_index ASC
    `,
    [messageId],
  );

  return rows;
}

async function indexAttachmentMetadata(
  mailboxId: number,
  messageId: number,
) {
  const pool = getDbPool();
  const connection = await pool.getConnection();

  try {
    await connection.beginTransaction();
    const [stateRows] = await connection.query<
      (RowDataPacket & Pick<MessageAttachmentStateRow, "attachments_indexed_at">)[]
    >(
      `
        SELECT attachments_indexed_at
        FROM mailbox_messages
        WHERE mailbox_id = ?
          AND id = ?
        LIMIT 1
        FOR UPDATE
      `,
      [mailboxId, messageId],
    );
    const state = stateRows[0];

    if (!state) {
      await connection.commit();
      return null;
    }

    if (state.attachments_indexed_at) {
      await connection.commit();
      return getAttachmentRows(messageId);
    }

    const [sourceRows] = await connection.query<
      (RowDataPacket & Pick<MessageAttachmentStateRow, "raw_source">)[]
    >(
      `
        SELECT raw_source
        FROM mailbox_messages
        WHERE mailbox_id = ?
          AND id = ?
        LIMIT 1
      `,
      [mailboxId, messageId],
    );
    const rawSource = sourceRows[0]?.raw_source ?? null;
    const attachments = rawSource
      ? await extractAttachmentsFromRawSource(rawSource)
      : [];

    await enqueueMailboxAttachmentStorageDeletesByMessageIds(connection, [messageId]);
    await connection.query(
      "DELETE FROM mailbox_message_attachments WHERE mailbox_message_id = ?",
      [messageId],
    );

    for (const attachment of attachments) {
      await connection.query(
        `
          INSERT INTO mailbox_message_attachments (
            mailbox_message_id,
            attachment_index,
            content_disposition,
            content_id,
            original_name,
            mime_type,
            size_bytes,
            storage_status
          ) VALUES (?, ?, ?, ?, ?, ?, ?, 'metadata')
        `,
        [
          messageId,
          attachment.index,
          attachment.contentDisposition,
          attachment.contentId,
          attachment.filename,
          attachment.contentType,
          attachment.sizeBytes,
        ],
      );
    }

    await connection.query(
      `
        UPDATE mailbox_messages
        SET attachments_indexed_at = NOW()
        WHERE mailbox_id = ?
          AND id = ?
      `,
      [mailboxId, messageId],
    );
    await connection.commit();
  } catch (error) {
    await connection.rollback();
    throw error;
  } finally {
    connection.release();
  }

  return getAttachmentRows(messageId);
}

export async function getMailboxMessageAttachmentMetadata(input: {
  mailboxId: number;
  messageId: number;
}) {
  await ensureOfficialMailSchema();
  const existing = attachmentMetadataQueues.get(input.messageId);

  if (existing) {
    return existing;
  }

  const current = indexAttachmentMetadata(input.mailboxId, input.messageId)
    .then((rows) => rows?.map(rowToMetadata) ?? [])
    .finally(() => {
      if (attachmentMetadataQueues.get(input.messageId) === current) {
        attachmentMetadataQueues.delete(input.messageId);
      }
    });
  attachmentMetadataQueues.set(input.messageId, current);
  return current;
}

export async function getCachedMailboxMessageAttachmentMetadata(input: {
  mailboxId: number;
  messageId: number;
}) {
  await ensureOfficialMailSchema();
  const [rows] = await getDbPool().query<AttachmentCacheRow[]>(
    `
      SELECT
        mma.attachment_index,
        mma.content_disposition,
        mma.content_id,
        mma.original_name,
        mma.mime_type,
        mma.size_bytes,
        mma.storage_status,
        mma.storage_key,
        mma.content_sha256
      FROM mailbox_message_attachments mma
      INNER JOIN mailbox_messages mm ON mm.id = mma.mailbox_message_id
      WHERE mma.mailbox_message_id = ?
        AND mm.mailbox_id = ?
      ORDER BY mma.attachment_index ASC
    `,
    [input.messageId, input.mailboxId],
  );

  return rows.map(rowToMetadata);
}

async function cacheMessageAttachmentPayloads(input: {
  mailboxId: number;
  messageId: number;
}) {
  const pool = getDbPool();
  const [messageRows] = await pool.query<
    (RowDataPacket & { raw_source: string | null })[]
  >(
    `
      SELECT raw_source
      FROM mailbox_messages
      WHERE mailbox_id = ?
        AND id = ?
      LIMIT 1
    `,
    [input.mailboxId, input.messageId],
  );
  const rawSource = messageRows[0]?.raw_source;

  if (!rawSource) {
    return [];
  }

  const payloads = await extractAttachmentPayloadsFromRawSource(rawSource);
  const [ownerRows] = await pool.query<AttachmentOwnerRow[]>(
    `
      SELECT
        u.id AS owner_user_id,
        LOWER(u.email) AS owner_email,
        mb.id AS mailbox_id,
        LOWER(mb.email) AS mailbox_email
      FROM mailbox_messages mm
      INNER JOIN mailboxes mb ON mb.id = mm.mailbox_id
      INNER JOIN users u ON u.id = mb.user_id
      WHERE mm.id = ?
        AND mm.mailbox_id = ?
      LIMIT 1
    `,
    [input.messageId, input.mailboxId],
  );
  const owner = ownerRows[0];

  if (!owner) return payloads;

  for (const payload of payloads) {
    const checksum = createHash("sha256").update(payload.content).digest("hex");
    const storageKey = [
      "official-mail",
      "mailbox-attachments",
      String(input.mailboxId),
      String(input.messageId),
      `${payload.index}-${checksum}`,
    ].join("/");

    try {
      const reservationConnection = await pool.getConnection();
      let attachmentRowId = 0;
      let alreadyCached = false;
      try {
        await reservationConnection.beginTransaction();
        await reservationConnection.query(
          `
            INSERT INTO mailbox_message_attachments (
              mailbox_message_id,
              attachment_index,
              content_disposition,
              content_id,
              original_name,
              mime_type,
              size_bytes,
              storage_status
            ) VALUES (?, ?, ?, ?, ?, ?, ?, 'metadata')
            ON DUPLICATE KEY UPDATE
              content_disposition = VALUES(content_disposition),
              content_id = VALUES(content_id),
              original_name = VALUES(original_name),
              mime_type = VALUES(mime_type),
              size_bytes = VALUES(size_bytes),
              updated_at = NOW()
          `,
          [
            input.messageId,
            payload.index,
            payload.contentDisposition,
            payload.contentId,
            payload.filename,
            payload.contentType,
            payload.sizeBytes,
          ],
        );
        const [reservationRows] = await reservationConnection.query<AttachmentReservationRow[]>(
          `
            SELECT id, storage_status, storage_key
            FROM mailbox_message_attachments
            WHERE mailbox_message_id = ? AND attachment_index = ?
            LIMIT 1
            FOR UPDATE
          `,
          [input.messageId, payload.index],
        );
        const reservation = reservationRows[0];
        if (!reservation) throw new Error("attachment-cache-reservation-missing");
        attachmentRowId = Number(reservation.id);
        alreadyCached = reservation.storage_status === "cached" && reservation.storage_key === storageKey;

        if (!alreadyCached) {
          await reserveMailboxStorageObject(reservationConnection, {
            mailboxEmail: owner.mailbox_email,
            mailboxId: owner.mailbox_id,
            mimeType: payload.contentType,
            originalName: payload.filename,
            ownerEmail: owner.owner_email,
            ownerUserId: owner.owner_user_id,
            sizeBytes: payload.sizeBytes,
            sourceKind: "mail_attachment",
            sourceRef: String(attachmentRowId),
            storageKey,
          });
        }
        await reservationConnection.commit();
      } catch (error) {
        await reservationConnection.rollback();
        throw error;
      } finally {
        reservationConnection.release();
      }

      if (alreadyCached) continue;

      const uploaded = await uploadPrivateBufferToObjectStorage({
        body: payload.content,
        cacheControl: "private, max-age=31536000, immutable",
        contentType: payload.contentType,
        key: storageKey,
      });

      const activationConnection = await pool.getConnection();
      try {
        await activationConnection.beginTransaction();
        const [activated] = await activationConnection.query<ResultSetHeader>(
          `
            UPDATE mailbox_message_attachments
            SET storage_status = 'cached',
                storage_bucket = ?,
                storage_key = ?,
                content_sha256 = ?,
                last_error = NULL,
                cached_at = NOW(),
                updated_at = NOW()
            WHERE id = ?
          `,
          [uploaded.bucketName, uploaded.key, checksum, attachmentRowId],
        );
        if (activated.affectedRows !== 1) throw new Error("attachment-cache-reservation-lost");
        await activateMailboxStorageObject(activationConnection, {
          storageBucket: uploaded.bucketName,
          storageKey: uploaded.key,
        });
        await activationConnection.commit();
      } catch (error) {
        await activationConnection.rollback();
        throw error;
      } finally {
        activationConnection.release();
      }
    } catch (error) {
      const errorMessage = error instanceof Error ? error.message : "attachment-cache-failed";
      const cleanupConnection = await pool.getConnection();
      try {
        await cleanupConnection.beginTransaction();
        const [reservationRows] = await cleanupConnection.query<AttachmentReservationRow[]>(
          `
            SELECT id, storage_status, storage_key
            FROM mailbox_message_attachments
            WHERE mailbox_message_id = ? AND attachment_index = ?
            LIMIT 1
            FOR UPDATE
          `,
          [input.messageId, payload.index],
        );
        const reservation = reservationRows[0];
        if (reservation && reservation.storage_status !== "cached") {
          await cleanupConnection.query(
            `UPDATE mailbox_message_attachments SET storage_status = 'failed', last_error = ?, updated_at = NOW() WHERE id = ?`,
            [errorMessage.slice(0, 1000), reservation.id],
          );
          await scheduleMailboxStorageObjectDeletion(cleanupConnection, {
            ownerUserId: owner.owner_user_id,
            reason: "attachment-cache-failed",
            sourceKind: "mail_attachment",
            sourceRef: String(reservation.id),
            storageKey,
          });
        }
        await cleanupConnection.commit();
      } catch {
        await cleanupConnection.rollback();
      } finally {
        cleanupConnection.release();
      }
    }
  }

  return payloads;
}

async function getOrCreateMessageAttachmentPayloads(input: {
  mailboxId: number;
  messageId: number;
}) {
  const existing = attachmentCacheQueues.get(input.messageId);

  if (existing) {
    return existing;
  }

  const current = cacheMessageAttachmentPayloads(input).finally(() => {
    if (attachmentCacheQueues.get(input.messageId) === current) {
      attachmentCacheQueues.delete(input.messageId);
    }
  });
  attachmentCacheQueues.set(input.messageId, current);
  return current;
}

export async function getCachedMailboxAttachmentPayload(input: {
  attachmentIndex: number;
  mailboxId: number;
  messageId: number;
}): Promise<RemoteMailboxAttachmentPayload | null> {
  await ensureOfficialMailSchema();
  const rows = await indexAttachmentMetadata(input.mailboxId, input.messageId);

  if (!rows) {
    return null;
  }

  const row = rows.find(
    (candidate) => Number(candidate.attachment_index) === input.attachmentIndex,
  );

  if (!row) {
    return null;
  }

  if (row.storage_status === "cached" && row.storage_key?.trim()) {
    try {
      const storedObject = await readObjectFromObjectStorage({
        key: row.storage_key.trim(),
      });

      if (storedObject?.body) {
        return {
          ...rowToMetadata(row),
          content: storedObject.body,
          contentType: storedObject.contentType?.trim() || row.mime_type,
        };
      }
    } catch {
      // Rebuild the private cache from the immutable MIME source below.
    }
  }

  const payloads = await getOrCreateMessageAttachmentPayloads(input);
  return payloads.find((payload) => payload.index === input.attachmentIndex) ?? null;
}