egbim_qa_platform/scripts/migrate-secretary-comments-to-abc.ts

import { randomUUID } from 'node:crypto';
import { existsSync } from 'node:fs';
import { readFile } from 'node:fs/promises';
import { join, resolve } from 'node:path';
import { config as loadEnv } from 'dotenv';
import {
  DeleteObjectCommand,
  GetObjectCommand,
  PutObjectCommand,
  S3Client,
} from '@aws-sdk/client-s3';
import mysql, { type Connection, type RowDataPacket } from 'mysql2/promise';

loadEnv({ path: resolve(__dirname, '../apps/secretary-api/.env') });

interface CommentRow extends RowDataPacket {
  source_comment_id: number;
  feedback_id: string;
  channel_id: string;
  author_id: string;
  author_tenant_id: string;
  author_name: string | null;
  content: string;
  is_internal: number;
  created_at: Date;
  updated_at: Date;
}

interface AttachmentRow extends RowDataPacket {
  source_attachment_id: number;
  source_comment_id: number;
  channel_id: string;
  storage_provider: string;
  storage_bucket: string | null;
  storage_key: string;
  original_file_name: string;
  mime_type: string | null;
  file_size: number | null;
}

interface ChannelRow extends RowDataPacket {
  id: number;
  image_config: unknown;
}

interface ImageConfig {
  accessKeyId: string;
  secretAccessKey: string;
  endpoint: string;
  region: string;
  bucket: string;
}

interface StoredObject {
  body: Buffer;
  contentType: string;
}

const isDryRun = !process.argv.includes('--apply');
const sourceLocalRoot = '/home/b24014/baron_qa/uploads';

const parseImageConfig = (raw: unknown): ImageConfig => {
  const value = typeof raw === 'string' ? JSON.parse(raw) : raw;
  if (!value || typeof value !== 'object') {
    throw new Error('ABC channel image_config is missing');
  }
  const config = value as Record<string, unknown>;
  const required = ['accessKeyId', 'secretAccessKey', 'endpoint', 'region', 'bucket'];
  if (required.some((key) => typeof config[key] !== 'string' || !config[key])) {
    throw new Error('ABC channel image_config is incomplete');
  }
  return config as unknown as ImageConfig;
};

const createClient = (config: ImageConfig) =>
  new S3Client({
    credentials: {
      accessKeyId: config.accessKeyId,

      secretAccessKey: config.secretAccessKey,
    },
    endpoint: config.endpoint,
    region: config.region,
  });

const connectionOptions = (rawUrl: string) => {
  const url = new URL(rawUrl.replace('mysql+pymysql://', 'mysql://'));
  return {
    host: url.hostname,
    port: Number(url.port || 3306),
    user: decodeURIComponent(url.username),
    password: decodeURIComponent(url.password),
    database: url.pathname.slice(1),
  };
};

const getSourceR2Client = () => {
  const endpoint = process.env.R2_ENDPOINT;
  const accessKeyId = process.env.R2_ACCESS_KEY_ID;
  const secretAccessKey = process.env.R2_SECRET_ACCESS_KEY;
  const region = process.env.R2_REGION ?? 'auto';
  if (!endpoint || !accessKeyId || !secretAccessKey) {
    throw new Error('Secretary R2 environment variables are incomplete');
  }
  return new S3Client({
    credentials: { accessKeyId, secretAccessKey },
    endpoint,
    region,
  });
};

const readSourceObject = async (attachment: AttachmentRow): Promise<StoredObject> => {
  const provider = attachment.storage_provider.toUpperCase();
  if (provider === 'LOCAL') {
    const root = resolve(attachment.storage_bucket ?? sourceLocalRoot);
    const path = resolve(join(root, attachment.storage_key));
    if (!path.startsWith(root + '/')) {
      throw new Error(`Unsafe local attachment path: ${attachment.storage_key}`);
    }
    if (!existsSync(path)) {
      throw new Error(`Local attachment not found: ${path}`);
    }
    return {
      body: await readFile(path),
      contentType: attachment.mime_type ?? 'application/octet-stream',
    };
  }

  if (provider !== 'R2') {
    throw new Error(`Unsupported Secretary storage provider: ${attachment.storage_provider}`);
  }
  const bucket = attachment.storage_bucket ?? process.env.R2_BUCKET;
  if (!bucket) {
    throw new Error(`Secretary R2 bucket is missing for attachment ${attachment.source_attachment_id}`);
  }
  const response = await getSourceR2Client().send(
    new GetObjectCommand({ Bucket: bucket, Key: attachment.storage_key }),
  );
  if (!response.Body || typeof response.Body.transformToByteArray !== 'function') {
    throw new Error(`Secretary R2 object has no readable body: ${attachment.storage_key}`);
  }
  return {
    body: Buffer.from(await response.Body.transformToByteArray()),
    contentType: attachment.mime_type ?? 'application/octet-stream',
  };
};

const findExistingComment = async (db: Connection, comment: CommentRow) => {
  const [rows] = await db.query<RowDataPacket[]>(
    `SELECT id FROM feedback_comments
      WHERE feedback_id = ? AND author_id = ? AND author_tenant_id = ?
        AND content = ? AND is_internal = ? AND created_at = ?
        AND deleted_at IS NULL LIMIT 1`,
    [
      Number(comment.feedback_id),
      comment.author_id,
      comment.author_tenant_id,
      comment.content,
      comment.is_internal ? 1 : 0,
      comment.created_at,
    ],
  );
  return rows[0] ? Number(rows[0].id) : null;
};

const insertComment = async (db: Connection, comment: CommentRow) => {
  const existingId = await findExistingComment(db, comment);
  if (existingId) return { id: existingId, created: false };

  const [result] = await db.query<mysql.ResultSetHeader>(
    `INSERT INTO feedback_comments
      (feedback_id, author_id, author_tenant_id, author_name, content,
       is_internal, comment_type, created_at, updated_at)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
    [
      Number(comment.feedback_id),
      comment.author_id,
      comment.author_tenant_id,
      comment.author_name ?? comment.author_id,
      comment.content,
      comment.is_internal ? 1 : 0,
      comment.is_internal ? 'INTERNAL_MEMO' : 'COMMENT',
      comment.created_at,
      comment.updated_at,
    ],
  );
  return { id: result.insertId, created: true };
};

const findExistingAttachment = async (
  db: Connection,
  commentId: number,
  attachment: AttachmentRow,
) => {
  const [rows] = await db.query<RowDataPacket[]>(
    `SELECT id FROM feedback_comment_attachments
      WHERE comment_id = ? AND original_file_name = ? AND mime_type = ?
        AND file_size = ? AND deleted_at IS NULL LIMIT 1`,
    [
      commentId,
      attachment.original_file_name,
      attachment.mime_type ?? 'application/octet-stream',
      attachment.file_size ?? 0,
    ],
  );
  return rows[0] ? Number(rows[0].id) : null;
};

const insertAttachment = async (
  db: Connection,
  config: ImageConfig,
  attachment: AttachmentRow,
  commentId: number,
) => {
  const existingId = await findExistingAttachment(db, commentId, attachment);
  if (existingId) return { id: existingId, created: false };

  const object = await readSourceObject(attachment);
  const safeName = attachment.original_file_name.replace(/[^a-zA-Z0-9._-]/g, '_');
  const storageKey =
    'migrated/secretary-comments/feedback-' +
    attachment.channel_id +
    '/comment-' +
    String(commentId) +
    '/' +
    randomUUID() +
    '-' +
    safeName;
  const client = createClient(config);

  await client.send(
    new PutObjectCommand({
      Bucket: config.bucket,
      Key: storageKey,
      Body: object.body,
      ContentType: object.contentType,
    }),
  );

  try {
    const [result] = await db.query<mysql.ResultSetHeader>(
      `INSERT INTO feedback_comment_attachments
        (comment_id, original_file_name, storage_key, storage_bucket,
         mime_type, file_size, created_at, updated_at)
       VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`,
      [
        commentId,
        attachment.original_file_name,
        storageKey,
        config.bucket,
        object.contentType,
        object.body.byteLength,
      ],
    );
    return { id: result.insertId, created: true };
  } catch (error) {
    await client
      .send(new DeleteObjectCommand({ Bucket: config.bucket, Key: storageKey }))
      .catch(() => undefined);
    throw error;
  }
};

const main = async () => {
  const secretaryUrl = process.env.DATABASE_URL;
  if (!secretaryUrl) throw new Error('Secretary DATABASE_URL is missing');
  const secretary = await mysql.createConnection(connectionOptions(secretaryUrl));
  const abc = await mysql.createConnection(
    connectionOptions('mysql://userfeedback:userfeedback@127.0.0.1:13306/userfeedback'),
  );

  try {
    const [comments] = await secretary.query<CommentRow[]>(
      `SELECT c.id AS source_comment_id, m.abc_feedback_id AS feedback_id,
              m.abc_channel_id AS channel_id, c.author_id, c.author_tenant_id,
              c.author_name, c.content, c.is_internal, c.created_at, c.updated_at
         FROM ticket_comments c
         JOIN abc_feedback_mappings m ON m.ticket_id = c.ticket_id
        WHERE c.deleted_at IS NULL AND m.abc_feedback_id IS NOT NULL
        ORDER BY c.id`,
    );
    const [attachments] = await secretary.query<AttachmentRow[]>(
      `SELECT a.id AS source_attachment_id, a.comment_id AS source_comment_id,
              m.abc_channel_id AS channel_id, a.storage_provider,
              a.storage_bucket, a.storage_key, a.original_file_name,
              a.mime_type, a.file_size
         FROM attachments a
         JOIN ticket_comments c ON c.id = a.comment_id AND c.deleted_at IS NULL
         JOIN abc_feedback_mappings m ON m.ticket_id = c.ticket_id
        WHERE a.deleted_at IS NULL AND a.comment_id IS NOT NULL
          AND m.abc_feedback_id IS NOT NULL
        ORDER BY a.id`,
    );

    const channelIds = [...new Set(attachments.map((item) => Number(item.channel_id)))];
    const configs = new Map<number, ImageConfig>();
    if (channelIds.length > 0) {
      const [channels] = await abc.query<ChannelRow[]>(
        'SELECT id, image_config FROM channels WHERE id IN (?)',
        [channelIds],
      );
      for (const channel of channels) {
        configs.set(channel.id, parseImageConfig(channel.image_config));
      }
    }

    console.log(`${isDryRun ? 'DRY-RUN' : 'APPLY'} comments: ${comments.length}`);
    console.log(`${isDryRun ? 'DRY-RUN' : 'APPLY'} comment attachments: ${attachments.length}`);

    const targetCommentIds = new Map<number, number>();
    let createdComments = 0;
    let skippedComments = 0;
    for (const comment of comments) {
      const existingId = await findExistingComment(abc, comment);
      if (isDryRun) {
        if (existingId) skippedComments += 1;
        else createdComments += 1;
        if (existingId) targetCommentIds.set(comment.source_comment_id, existingId);
        continue;
      }
      const result = await insertComment(abc, comment);
      targetCommentIds.set(comment.source_comment_id, result.id);
      result.created ? (createdComments += 1) : (skippedComments += 1);
    }

    let createdAttachments = 0;
    let skippedAttachments = 0;
    let pendingAttachments = 0;
    for (const attachment of attachments) {
      const commentId = targetCommentIds.get(attachment.source_comment_id);
      if (!commentId) {
        pendingAttachments += 1;
        continue;
      }
      const config = configs.get(Number(attachment.channel_id));
      if (!config) throw new Error(`ABC channel ${attachment.channel_id} R2 config is missing`);
      const existingId = await findExistingAttachment(abc, commentId, attachment);
      if (isDryRun) {
        existingId ? (skippedAttachments += 1) : (createdAttachments += 1);
        continue;
      }
      const result = await insertAttachment(abc, config, attachment, commentId);
      result.created ? (createdAttachments += 1) : (skippedAttachments += 1);
    }

    console.log('Comments created:', createdComments);
    console.log('Comments skipped:', skippedComments);
    console.log('Attachments created:', createdAttachments);
    console.log('Attachments skipped:', skippedAttachments);
    console.log('Attachments pending:', pendingAttachments);
    if (isDryRun) console.log('No database or R2 changes were made. Use --apply to execute.');
  } finally {
    await secretary.end();
    await abc.end();
  }
};

main().catch((error) => {
  console.error(error instanceof Error ? error.message : error);
  process.exitCode = 1;
});