src/db.js

// 多数据库工具 (PostgreSQL + MongoDB + MySQL)
import fs from "fs";

import pkg from "pg";
const { Pool } = pkg;
import { MongoClient } from "mongodb";
import mysql from "mysql2/promise";
import dotenv from "dotenv";
dotenv.config();

// Utility function to format error information for better logging
function formatErrorInfo(error) {
  if (!error) return { errorMsg: '未知错误', errorCode: '无错误代码' };
  
  let errorMsg = error?.message || error?.toString() || '未知错误';
  const errorCode = error?.code || '无错误代码';
  
  // Handle AggregateError specially
  if (error instanceof AggregateError && error.errors?.length > 0) {
    const innerError = error.errors[0];
    const innerMsg = innerError?.message || innerError?.toString() || '内部错误';
    errorMsg = `${errorMsg} (${innerMsg})`;
  }
  
  return { errorMsg, errorCode };
}
if (fs.existsSync(".env.local")) {
  console.log("Using .env.local file to supply config environment variables");
  const envConfig = dotenv.parse(fs.readFileSync(".env.local"));
  for (const k in envConfig) {
    process.env[k] = envConfig[k];
  }
} else {
  console.log(
    "Using .env file to supply config environment variables, you can create a .env.local file to overwrite defaults, it doesn't upload to git"
  );
}
// 主数据库连接池 (Aiven PostgreSQL)
const pool = new Pool({
  connectionString: process.env.POSTGRES_URI,
  max: 5,
  idleTimeoutMillis: 30000,
  connectionTimeoutMillis: 2000,
  ssl: { rejectUnauthorized: false },
});

// 备用数据库连接池 (CockroachDB)
const cockroachPool = new Pool({
  connectionString: process.env.COCKROACH_URI,
  max: 3,
  idleTimeoutMillis: 30000,
  connectionTimeoutMillis: 2000,
  ssl: { rejectUnauthorized: false },
});

// 备用数据库连接池 (Neon)
// 备用数据库连接池 (Neon)
const neonPool = new Pool({
  connectionString: process.env.NEON_URI,
  max: 3,
  idleTimeoutMillis: 30000,
  connectionTimeoutMillis: 2000,
  ssl: { rejectUnauthorized: false },
});

// MySQL 连接池 (Aiven MySQL)
let mysqlPool;

// 初始化 MySQL 连接池
async function initMySQL() {
  if (process.env.AIVEN_MYSQL_URI && !mysqlPool) {
    try {
      mysqlPool = mysql.createPool({
        uri: process.env.AIVEN_MYSQL_URI,
        connectionLimit: 5,
        acquireTimeout: 60000,
        timeout: 60000,
        ssl: { rejectUnauthorized: false },
      });
      console.log("✅ MySQL 连接池创建成功");
    } catch (error) {
      console.error("❌ MySQL 连接池创建失败:", error.message);
      mysqlPool = null;
    }
  }
  return mysqlPool;
}

// MongoDB 连接
let mongoClient;
let mongoDb;

// 初始化 MongoDB 连接
async function initMongoDB() {
  if (process.env.MONGO_URI && !mongoClient) {
    try {
      mongoClient = new MongoClient(process.env.MONGO_URI, {
        maxPoolSize: 5,
        serverSelectionTimeoutMS: 5000,
        socketTimeoutMS: 10000,
      });
      // 连接到数据库
      await mongoClient.connect();
      mongoDb = mongoClient.db("auto_read_posts"); // 使用专门的数据库名
      console.log("✅ MongoDB 连接成功");
    } catch (error) {
      console.error("❌ MongoDB 连接失败:", error.message);
      mongoClient = null;
      mongoDb = null;
    }
  }
  return mongoDb;
}

// 所有数据库连接池数组 (PostgreSQL)
const allPools = [
  { name: "Aiven PostgreSQL", pool: pool },
  { name: "CockroachDB", pool: cockroachPool },
  // { name: "Neon", pool: neonPool },
];

// 获取所有数据库连接数组 (包括 MongoDB 和 MySQL)
async function getAllDatabases() {
  const mongoDb = await initMongoDB();
  const mysqlPool = await initMySQL();

  return [
    ...allPools,
    ...(mongoDb ? [{ name: "MongoDB", db: mongoDb, type: "mongo" }] : []),
    ...(mysqlPool
      ? [{ name: "Aiven MySQL", pool: mysqlPool, type: "mysql" }]
      : []),
  ];
}

export async function savePosts(posts) {
  if (!Array.isArray(posts) || posts.length === 0) {
    console.warn("无效的帖子数据或空数组，跳过保存");
    return;
  }

  // 验证帖子数据
  const validPosts = posts.filter(post => {
    if (!post || !post.guid || typeof post.guid !== 'string' || post.guid.trim() === '') {
      console.warn(`跳过无效帖子数据: ${JSON.stringify({title: post?.title, guid: post?.guid})}`);
      return false;
    }
    return true;
  });

  if (validPosts.length === 0) {
    console.warn("没有有效的帖子数据，跳过保存");
    return;
  }

  const allDatabases = await getAllDatabases();

  if (allDatabases.length === 0) {
    console.warn("没有可用的数据库连接，跳过保存");
    return;
  }

  const createTableQuery = `
    CREATE TABLE IF NOT EXISTS posts (
      id SERIAL PRIMARY KEY,
      title TEXT,
      creator TEXT,
      description TEXT,
      link TEXT,
      pubDate TEXT,
      guid TEXT UNIQUE,
      guidIsPermaLink TEXT,
      source TEXT,
      sourceUrl TEXT,
      created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
  `;

  const insertQuery = `
    INSERT INTO posts (title, creator, description, link, pubDate, guid, guidIsPermaLink, source, sourceUrl)
    VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
    ON CONFLICT (guid) DO NOTHING
  `;
  // 并行操作所有数据库
  const savePromises = allDatabases.map(async ({ name, pool, db, type }) => {
    try {
      console.log(`正在保存到 ${name}...`);

      if (type === "mongo" && db) {
        // MongoDB 操作
        const collection = db.collection("posts");

        // 准备 MongoDB 文档
        const mongoDocuments = validPosts.map((post) => ({
          title: post.title,
          creator: post.creator,
          description: post.description,
          link: post.link,
          pubDate: post.pubDate,
          guid: post.guid,
          guidIsPermaLink: post.guidIsPermaLink,
          source: post.source,
          sourceUrl: post.sourceUrl,
          created_at: new Date(),
        }));

        // 使用 upsert 操作避免重复
        const bulkOps = mongoDocuments.map((doc) => ({
          updateOne: {
            filter: { guid: doc.guid },
            update: { $set: doc },
            upsert: true,
          },
        }));
        if (bulkOps.length > 0) {
          await collection.bulkWrite(bulkOps);
        }
      } else if (type === "mysql" && pool) {
        // MySQL 操作
        const mysqlCreateTableQuery = `
          CREATE TABLE IF NOT EXISTS posts (
            id INT AUTO_INCREMENT PRIMARY KEY,
            title TEXT,
            creator TEXT,
            description TEXT,
            link TEXT,
            pubDate TEXT,
            guid VARCHAR(500) UNIQUE,
            guidIsPermaLink TEXT,
            source TEXT,
            sourceUrl TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
          )
        `;

        const mysqlInsertQuery = `
          INSERT INTO posts (title, creator, description, link, pubDate, guid, guidIsPermaLink, source, sourceUrl)
          VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
          ON DUPLICATE KEY UPDATE
          title = VALUES(title),
          creator = VALUES(creator),
          description = VALUES(description),
          link = VALUES(link),
          pubDate = VALUES(pubDate),
          guidIsPermaLink = VALUES(guidIsPermaLink),
          source = VALUES(source),
          sourceUrl = VALUES(sourceUrl)
        `;

        // 建表
        await pool.execute(mysqlCreateTableQuery);

        // 插入数据
        for (const post of validPosts) {
          await pool.execute(mysqlInsertQuery, [
            post.title,
            post.creator,
            post.description,
            post.link,
            post.pubDate,
            post.guid,
            post.guidIsPermaLink,
            post.source,
            post.sourceUrl,
          ]);
        }
      } else if (pool) {
        // PostgreSQL 操作
        // 建表
        await pool.query(createTableQuery);

        // 插入数据
        for (const post of validPosts) {
          await pool.query(insertQuery, [
            post.title,
            post.creator,
            post.description,
            post.link,
            post.pubDate,
            post.guid,
            post.guidIsPermaLink,
            post.source,
            post.sourceUrl,
          ]);
        }
      }

      console.log(`✅ ${name} 保存成功 (${validPosts.length} 条记录)`);
      return { name, success: true };
    } catch (error) {
      const { errorMsg, errorCode } = formatErrorInfo(error);
      console.error(`❌ ${name} 保存失败 [${errorCode}]:`, errorMsg);
      return { name, success: false, error: errorMsg };
    }
  });

  // 等待所有数据库操作完成
  const results = await Promise.allSettled(savePromises);

  // 统计结果
  const successCount = results.filter(
    (result) => result.status === "fulfilled" && result.value.success
  ).length;

  console.log(
    `数据库保存结果: ${successCount}/${allDatabases.length} 个数据库保存成功`
  );

  // 如果至少有一个数据库保存成功，就认为操作成功
  if (successCount === 0) {
    throw new Error("所有数据库保存都失败了");
  }
}

export async function isGuidExists(guid) {
  // 验证输入参数
  if (!guid || typeof guid !== 'string' || guid.trim() === '') {
    console.warn(`无效的GUID参数: ${JSON.stringify(guid)}`);
    return false;
  }

  // 优先查询主数据库 (Aiven PostgreSQL)
  try {
    const res = await pool.query(
      "SELECT 1 FROM posts WHERE guid = $1 LIMIT 1",
      [guid]
    );
    // console.log("isGuidExists查询结果:", res.rows); 存在的返回[ { '?column?': 1 } ]
    if (res.rowCount > 0) {
      return true;
    }
  } catch (error) {
    const { errorMsg, errorCode } = formatErrorInfo(error);
    console.warn(`主数据库查询GUID失败 [${errorCode}]: ${errorMsg}`);
  }
  // 如果主数据库查询失败或未找到，尝试查询备用数据库
  const allDatabases = await getAllDatabases();
  for (const { name, pool, db, type } of allDatabases.slice(1)) {
    // 跳过主数据库
    try {
      if (type === "mongo" && db) {
        // MongoDB 查询
        const collection = db.collection("posts");
        const count = await collection.countDocuments(
          { guid: guid },
          { limit: 1 }
        );
        if (count > 0) {
          console.log(`在备用数据库 ${name} 中找到GUID: ${guid}`);
          return true;
        }
      } else if (type === "mysql" && pool) {
        // MySQL 查询
        const [rows] = await pool.execute(
          "SELECT 1 FROM posts WHERE guid = ? LIMIT 1",
          [guid]
        );
        if (rows.length > 0) {
          console.log(`在备用数据库 ${name} 中找到GUID: ${guid}`);
          return true;
        }
      } else if (pool) {
        // PostgreSQL 查询
        const res = await pool.query(
          "SELECT 1 FROM posts WHERE guid = $1 LIMIT 1",
          [guid]
        );
        if (res.rowCount > 0) {
          console.log(`在备用数据库 ${name} 中找到GUID: ${guid}`);
          return true;
        }
      }
    } catch (error) {
      const { errorMsg, errorCode } = formatErrorInfo(error);
      console.warn(`备用数据库 ${name} 查询GUID失败 [${errorCode}]: ${errorMsg}`);
    }
  }

  return false;
}

// 测试所有数据库连接
export async function testAllConnections() {
  console.log("正在测试所有数据库连接...");
  const allDatabases = await getAllDatabases();
  const testPromises = allDatabases.map(async ({ name, pool, db, type }) => {
    try {
      if (type === "mongo" && db) {
        // 测试 MongoDB 连接
        await db.admin().ping();
      } else if (type === "mysql" && pool) {
        // 测试 MySQL 连接
        await pool.execute("SELECT 1");
      } else if (pool) {
        // 测试 PostgreSQL 连接
        await pool.query("SELECT 1");
      }
      console.log(`✅ ${name} 连接正常`);
      return { name, connected: true };
    } catch (error) {
      const { errorMsg, errorCode } = formatErrorInfo(error);
      console.error(`❌ ${name} 连接失败 [${errorCode}]:`, errorMsg);
      return { name, connected: false, error: errorMsg };
    }
  });

  const results = await Promise.allSettled(testPromises);
  const connectedCount = results.filter(
    (result) => result.status === "fulfilled" && result.value.connected
  ).length;

  console.log(
    `数据库连接测试结果: ${connectedCount}/${allDatabases.length} 个数据库连接正常`
  );
  return results;
}

// 获取所有数据库的统计信息
export async function getAllDatabaseStats() {
  console.log("正在获取所有数据库统计信息...");
  const allDatabases = await getAllDatabases();
  const statsPromises = allDatabases.map(async ({ name, pool, db, type }) => {
    try {
      let stats;

      if (type === "mongo" && db) {
        // MongoDB 统计
        const collection = db.collection("posts");
        const totalPosts = await collection.countDocuments();
        const latestPost = await collection.findOne(
          {},
          { sort: { created_at: -1 } }
        );
        stats = {
          name,
          totalPosts,
          latestPost: latestPost?.created_at || null,
          status: "healthy",
        };
      } else if (type === "mysql" && pool) {
        // MySQL 统计
        const [countResult] = await pool.execute(
          "SELECT COUNT(*) as count FROM posts"
        );
        const [latestResult] = await pool.execute(
          "SELECT created_at FROM posts ORDER BY created_at DESC LIMIT 1"
        );

        stats = {
          name,
          totalPosts: parseInt(countResult[0].count),
          latestPost: latestResult[0]?.created_at || null,
          status: "healthy",
        };
      } else if (pool) {
        // PostgreSQL 统计
        const countResult = await pool.query(
          "SELECT COUNT(*) as count FROM posts"
        );
        const latestResult = await pool.query(
          "SELECT created_at FROM posts ORDER BY created_at DESC LIMIT 1"
        );

        stats = {
          name,
          totalPosts: parseInt(countResult.rows[0].count),
          latestPost: latestResult.rows[0]?.created_at || null,
          status: "healthy",
        };
      }

      console.log(`📊 ${name}: ${stats.totalPosts} 条记录`);
      return stats;
    } catch (error) {
      const { errorMsg, errorCode } = formatErrorInfo(error);
      console.error(`❌ ${name} 统计信息获取失败 [${errorCode}]:`, errorMsg);
      return {
        name,
        totalPosts: -1,
        latestPost: null,
        status: "error",
        error: errorMsg,
      };
    }
  });

  const results = await Promise.allSettled(statsPromises);
  return results.map((result) =>
    result.status === "fulfilled" ? result.value : result.reason
  );
}

// 关闭所有数据库连接
export async function closeAllConnections() {
  console.log("正在关闭所有数据库连接...");
  const allDatabases = await getAllDatabases();
  const closePromises = allDatabases.map(async ({ name, pool, type }) => {
    try {
      if (type === "mongo") {
        // 关闭 MongoDB 连接
        if (mongoClient) {
          await mongoClient.close();
          mongoClient = null;
          mongoDb = null;
        }
      } else if (type === "mysql" && pool) {
        // 关闭 MySQL 连接
        await pool.end();
        mysqlPool = null;
      } else if (pool) {
        // 关闭 PostgreSQL 连接
        await pool.end();
      }
      console.log(`✅ ${name} 连接已关闭`);
    } catch (error) {
      const { errorMsg, errorCode } = formatErrorInfo(error);
      console.error(`❌ ${name} 连接关闭失败 [${errorCode}]:`, errorMsg);
    }
  });

  await Promise.allSettled(closePromises);
  console.log("所有数据库连接关闭完成");
}

// 保存话题 JSON 数据的函数
export async function saveTopicData(topicData) {
  if (!topicData || !topicData.id) {
    console.warn("无效的话题数据，跳过保存");
    return;
  }

  const allDatabases = await getAllDatabases();

  const createTopicsTableQuery = `
    CREATE TABLE IF NOT EXISTS topics (
      id SERIAL PRIMARY KEY,
      topic_id INTEGER UNIQUE,
      title TEXT,
      slug TEXT,
      posts_count INTEGER,
      created_at TIMESTAMP,
      last_posted_at TIMESTAMP,
      views INTEGER,
      like_count INTEGER,
      category_id INTEGER,
      tags TEXT[],
      raw_data JSONB,
      saved_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
  `;

  const insertTopicQuery = `
    INSERT INTO topics (topic_id, title, slug, posts_count, created_at, last_posted_at, views, like_count, category_id, tags, raw_data)
    VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
    ON CONFLICT (topic_id) DO UPDATE SET
      title = EXCLUDED.title,
      slug = EXCLUDED.slug,
      posts_count = EXCLUDED.posts_count,
      last_posted_at = EXCLUDED.last_posted_at,
      views = EXCLUDED.views,
      like_count = EXCLUDED.like_count,
      tags = EXCLUDED.tags,
      raw_data = EXCLUDED.raw_data,
      saved_at = CURRENT_TIMESTAMP
  `;

  const savePromises = allDatabases.map(async ({ name, pool, db, type }) => {
    try {
      console.log(`正在保存话题数据到 ${name}...`);

      if (type === "mongo" && db) {
        // MongoDB 操作
        const collection = db.collection("topics");
        
        const mongoDocument = {
          topic_id: topicData.id,
          title: topicData.title,
          slug: topicData.slug,
          posts_count: topicData.posts_count,
          created_at: new Date(topicData.created_at),
          last_posted_at: topicData.last_posted_at ? new Date(topicData.last_posted_at) : null,
          views: topicData.views,
          like_count: topicData.like_count,
          category_id: topicData.category_id,
          tags: topicData.tags || [],
          raw_data: topicData,
          saved_at: new Date(),
        };

        await collection.updateOne(
          { topic_id: topicData.id },
          { $set: mongoDocument },
          { upsert: true }
        );
      } else if (type === "mysql" && pool) {        // MySQL 操作
        const mysqlCreateTableQuery = `
          CREATE TABLE IF NOT EXISTS topics (
            id INT AUTO_INCREMENT PRIMARY KEY,
            topic_id INT UNIQUE,
            title TEXT,
            slug TEXT,
            posts_count INT,
            created_at DATETIME,
            last_posted_at DATETIME NULL,
            views INT,
            like_count INT,
            category_id INT,
            tags JSON,
            raw_data JSON,
            saved_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
          )
        `;

        const mysqlInsertQuery = `
          INSERT INTO topics (topic_id, title, slug, posts_count, created_at, last_posted_at, views, like_count, category_id, tags, raw_data)
          VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
          ON DUPLICATE KEY UPDATE
          title = VALUES(title),
          slug = VALUES(slug),
          posts_count = VALUES(posts_count),
          last_posted_at = VALUES(last_posted_at),
          views = VALUES(views),
          like_count = VALUES(like_count),
          tags = VALUES(tags),
          raw_data = VALUES(raw_data),
          saved_at = CURRENT_TIMESTAMP
        `;        await pool.execute(mysqlCreateTableQuery);
        
        // 转换日期格式为 MySQL 兼容格式 (YYYY-MM-DD HH:MM:SS)
        const formatDateForMySQL = (dateString) => {
          if (!dateString) return null;
          const date = new Date(dateString);
          return date.toISOString().slice(0, 19).replace('T', ' ');
        };
        
        await pool.execute(mysqlInsertQuery, [
          topicData.id,
          topicData.title,
          topicData.slug,
          topicData.posts_count,
          formatDateForMySQL(topicData.created_at),
          formatDateForMySQL(topicData.last_posted_at),
          topicData.views,
          topicData.like_count,
          topicData.category_id,
          JSON.stringify(topicData.tags || []),
          JSON.stringify(topicData),
        ]);
      } else if (pool) {
        // PostgreSQL 操作
        await pool.query(createTopicsTableQuery);
        
        await pool.query(insertTopicQuery, [
          topicData.id,
          topicData.title,
          topicData.slug,
          topicData.posts_count,
          topicData.created_at,
          topicData.last_posted_at,
          topicData.views,
          topicData.like_count,
          topicData.category_id,
          topicData.tags || [],
          topicData,
        ]);
      }

      console.log(`✅ ${name} 话题数据保存成功 (话题ID: ${topicData.id})`);
      return { name, success: true };
    } catch (error) {
      const { errorMsg, errorCode } = formatErrorInfo(error);
      console.error(`❌ ${name} 话题数据保存失败 [${errorCode}]:`, errorMsg);
      return { name, success: false, error: errorMsg };
    }
  });

  const results = await Promise.allSettled(savePromises);
  const successCount = results.filter(
    (result) => result.status === "fulfilled" && result.value.success
  ).length;

  console.log(
    `话题数据保存结果: ${successCount}/${allDatabases.length} 个数据库保存成功`
  );

  if (successCount === 0) {
    throw new Error("所有数据库话题数据保存都失败了");
  }
}