memind/db.mjs

import fs from 'node:fs';
import mysql from 'mysql2/promise';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import { ensureDefaultSpaces } from './mindspace.mjs';

const __dirname = path.dirname(fileURLToPath(import.meta.url));

export function isDatabaseConfigured() {
  return Boolean(
    process.env.DATABASE_URL ||
      (process.env.MYSQL_HOST && process.env.MYSQL_DATABASE),
  );
}

// Pool size must scale with peak concurrent agent sessions: each session does
// several queries per turn (session node lookup + policy + provider + reconcile).
// The historical default of 10 starved at ~30 concurrent sessions. Tune via
// MYSQL_POOL_SIZE per portal instance × expected concurrency.
function resolvePoolOptions() {
  const connectionLimit = Math.max(1, Number(process.env.MYSQL_POOL_SIZE ?? 50));
  // queueLimit caps how many requests wait for a free connection before failing
  // fast, instead of piling up unboundedly under a connection-starvation storm.
  const queueLimit = Math.max(0, Number(process.env.MYSQL_POOL_QUEUE_LIMIT ?? 0));
  return { waitForConnections: true, connectionLimit, queueLimit };
}

export function createDbPool() {
  if (!isDatabaseConfigured()) {
    throw new Error('MySQL 未配置，请设置 DATABASE_URL 或 MYSQL_* 环境变量');
  }

  const poolOptions = resolvePoolOptions();

  if (process.env.DATABASE_URL) {
    return mysql.createPool({ uri: process.env.DATABASE_URL, ...poolOptions });
  }

  return mysql.createPool({
    host: process.env.MYSQL_HOST ?? 'localhost',
    port: Number(process.env.MYSQL_PORT ?? 3306),
    user: process.env.MYSQL_USER ?? 'boot',
    password: process.env.MYSQL_PASSWORD ?? '',
    database: process.env.MYSQL_DATABASE ?? 'tkmind',
    ...poolOptions,
  });
}

async function columnExists(pool, table, column) {
  const [rows] = await pool.query(
    `SELECT 1 FROM information_schema.COLUMNS
     WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME = ?
     LIMIT 1`,
    [table, column],
  );
  return rows.length > 0;
}

async function indexExists(pool, table, index) {
  const [rows] = await pool.query(
    `SELECT 1 FROM information_schema.STATISTICS
     WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND INDEX_NAME = ?
     LIMIT 1`,
    [table, index],
  );
  return rows.length > 0;
}

async function foreignKeyDeleteRule(pool, table, constraint) {
  const [rows] = await pool.query(
    `SELECT DELETE_RULE
     FROM information_schema.REFERENTIAL_CONSTRAINTS
     WHERE CONSTRAINT_SCHEMA = DATABASE()
       AND TABLE_NAME = ?
       AND CONSTRAINT_NAME = ?
     LIMIT 1`,
    [table, constraint],
  );
  return rows[0]?.DELETE_RULE ?? null;
}

async function ensureForeignKeyDeleteRule(
  pool,
  { table, constraint, column, referencedTable, referencedColumn = 'id', deleteRule },
) {
  const currentRule = await foreignKeyDeleteRule(pool, table, constraint);
  if (currentRule === deleteRule) return;
  if (currentRule) {
    await pool.query(`ALTER TABLE \`${table}\` DROP FOREIGN KEY \`${constraint}\``);
  }
  await pool.query(
    `ALTER TABLE \`${table}\`
     ADD CONSTRAINT \`${constraint}\`
       FOREIGN KEY (\`${column}\`) REFERENCES \`${referencedTable}\`(\`${referencedColumn}\`)
       ON DELETE ${deleteRule}`,
  );
}

export async function migrateSchema(pool) {
  const renames = [
    ['h5_user_sessions', 'goose_session_id', 'agent_session_id'],
    ['h5_session_billing_state', 'goose_session_id', 'agent_session_id'],
    ['h5_usage_records', 'goose_session_id', 'agent_session_id'],
  ];
  for (const [table, oldCol, newCol] of renames) {
    if (await columnExists(pool, table, oldCol)) {
      await pool.query(
        `ALTER TABLE \`${table}\` CHANGE \`${oldCol}\` \`${newCol}\` VARCHAR(128) NOT NULL`,
      );
    }
  }

  const userColumns = [
    ['slug', 'VARCHAR(64) NULL AFTER username'],
    ['email', 'VARCHAR(255) NULL AFTER slug'],
    [
      'password_algorithm',
      "VARCHAR(32) NOT NULL DEFAULT 'pbkdf2-sha512' AFTER password_hash",
    ],
    ['plan_type', "VARCHAR(32) NOT NULL DEFAULT 'free' AFTER status"],
  ];
  for (const [column, definition] of userColumns) {
    if (!(await columnExists(pool, 'h5_users', column))) {
      await pool.query(`ALTER TABLE h5_users ADD COLUMN \`${column}\` ${definition}`);
    }
  }

  await pool.query(`UPDATE h5_users SET slug = username WHERE slug IS NULL OR slug = ''`);
  if (!(await indexExists(pool, 'h5_users', 'uq_h5_users_slug'))) {
    await pool.query(`ALTER TABLE h5_users ADD UNIQUE KEY uq_h5_users_slug (slug)`);
  }
  if (!(await indexExists(pool, 'h5_users', 'uq_h5_users_email'))) {
    await pool.query(`ALTER TABLE h5_users ADD UNIQUE KEY uq_h5_users_email (email)`);
  }
  if (!(await indexExists(pool, 'h5_usage_records', 'uniq_h5_usage_request_id'))) {
    await pool.query(
      `ALTER TABLE h5_usage_records ADD UNIQUE KEY uniq_h5_usage_request_id (request_id)`,
    );
  }

  const assetForeignKeys = [
    {
      table: 'h5_assets',
      constraint: 'fk_h5_asset_category',
      column: 'category_id',
      referencedTable: 'h5_space_categories',
      deleteRule: 'CASCADE',
    },
    {
      table: 'h5_assets',
      constraint: 'fk_h5_asset_parent',
      column: 'parent_id',
      referencedTable: 'h5_assets',
      deleteRule: 'SET NULL',
    },
    {
      table: 'h5_asset_versions',
      constraint: 'fk_h5_asset_version_user',
      column: 'created_by',
      referencedTable: 'h5_users',
      deleteRule: 'CASCADE',
    },
    {
      table: 'h5_upload_sessions',
      constraint: 'fk_h5_upload_category',
      column: 'category_id',
      referencedTable: 'h5_space_categories',
      deleteRule: 'CASCADE',
    },
    {
      table: 'h5_page_versions',
      constraint: 'fk_h5_page_version_content',
      column: 'content_asset_id',
      referencedTable: 'h5_assets',
      deleteRule: 'CASCADE',
    },
  ];
  for (const foreignKey of assetForeignKeys) {
    await ensureForeignKeyDeleteRule(pool, foreignKey);
  }

  await pool.query(
    `ALTER TABLE h5_assets
     MODIFY source_type ENUM('upload', 'chat', 'agent', 'template', 'generated', 'workspace')
     NOT NULL DEFAULT 'upload'`,
  );

  const llmColumns = [
    ['provider_kind', "ENUM('builtin','custom') NOT NULL DEFAULT 'builtin' AFTER provider_id"],
    ['api_url', 'VARCHAR(512) NULL AFTER provider_kind'],
    ['base_path', 'VARCHAR(256) NULL AFTER api_url'],
    ['models_json', 'TEXT NULL AFTER base_path'],
    ['goosed_provider_id', 'VARCHAR(64) NULL AFTER models_json'],
    ['engine', "VARCHAR(32) NOT NULL DEFAULT 'openai' AFTER goosed_provider_id"],
    ['relay_provider', 'VARCHAR(64) NULL AFTER engine'],
    ['is_vision_selected', 'TINYINT(1) NOT NULL DEFAULT 0 AFTER is_selected'],
  ];
  for (const [column, definition] of llmColumns) {
    if (!(await columnExists(pool, 'h5_llm_provider_keys', column))) {
      await pool.query(`ALTER TABLE h5_llm_provider_keys ADD COLUMN \`${column}\` ${definition}`);
    }
  }

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_llm_executor_bindings (
      id CHAR(36) PRIMARY KEY,
      executor ENUM('goose', 'aider', 'openhands') NOT NULL,
      purpose VARCHAR(32) NOT NULL DEFAULT 'default',
      provider_key_id CHAR(36) NULL,
      model VARCHAR(128) NOT NULL,
      enabled TINYINT(1) NOT NULL DEFAULT 1,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      UNIQUE KEY uq_h5_llm_executor_binding (executor, purpose),
      KEY idx_h5_llm_executor_provider (provider_key_id),
      CONSTRAINT fk_h5_llm_executor_provider FOREIGN KEY (provider_key_id) REFERENCES h5_llm_provider_keys(id) ON DELETE SET NULL
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  // Shared experience store (etat C). Keep in sync with schema.sql.
  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_experience (
      id CHAR(36) PRIMARY KEY,
      scope VARCHAR(64) NOT NULL DEFAULT 'global',
      kind VARCHAR(32) NOT NULL DEFAULT 'lesson',
      title VARCHAR(255) NOT NULL,
      body MEDIUMTEXT NOT NULL,
      tags_json JSON NULL,
      source_session_id VARCHAR(128) NULL,
      source_user_id CHAR(36) NULL,
      use_count BIGINT NOT NULL DEFAULT 0,
      embedding JSON NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      KEY idx_h5_experience_scope_updated (scope, updated_at),
      FULLTEXT KEY ftx_h5_experience (title, body)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  // Per-user conversation memory: raw dialogue rows plus lightweight extracted
  // memory items. This is intentionally additive and can be disabled by env.
  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_conversation_messages (
      id CHAR(64) PRIMARY KEY,
      user_id CHAR(36) NOT NULL,
      agent_session_id VARCHAR(128) NOT NULL,
      message_key VARCHAR(128) NOT NULL,
      sequence_no INT NOT NULL DEFAULT 0,
      role VARCHAR(32) NOT NULL,
      text MEDIUMTEXT NOT NULL,
      raw_json LONGTEXT NULL,
      analyzed_at BIGINT NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      UNIQUE KEY uq_h5_conversation_message (agent_session_id, message_key),
      KEY idx_h5_conversation_user_created (user_id, created_at),
      KEY idx_h5_conversation_user_analyzed (user_id, analyzed_at),
      CONSTRAINT fk_h5_conversation_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE,
      CONSTRAINT fk_h5_conversation_session FOREIGN KEY (agent_session_id) REFERENCES h5_user_sessions(agent_session_id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_agent_runs (
      id CHAR(36) PRIMARY KEY,
      user_id CHAR(36) NOT NULL,
      agent_session_id VARCHAR(128) NULL,
      request_id VARCHAR(128) NOT NULL,
      status ENUM('queued', 'running', 'retryable', 'succeeded', 'failed') NOT NULL DEFAULT 'queued',
      attempts INT NOT NULL DEFAULT 0,
      user_message_json LONGTEXT NOT NULL,
      error_message TEXT NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      started_at BIGINT NULL,
      completed_at BIGINT NULL,
      UNIQUE KEY uq_h5_agent_run_request (user_id, request_id),
      KEY idx_h5_agent_run_user_status (user_id, status, updated_at),
      KEY idx_h5_agent_run_session (agent_session_id, updated_at),
      CONSTRAINT fk_h5_agent_run_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE,
      CONSTRAINT fk_h5_agent_run_session FOREIGN KEY (agent_session_id) REFERENCES h5_user_sessions(agent_session_id) ON DELETE SET NULL
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_agent_run_events (
      id CHAR(36) PRIMARY KEY,
      run_id CHAR(36) NOT NULL,
      event_type VARCHAR(64) NOT NULL,
      data_json JSON NULL,
      created_at BIGINT NOT NULL,
      KEY idx_h5_agent_run_event_run (run_id, created_at),
      CONSTRAINT fk_h5_agent_run_event_run FOREIGN KEY (run_id) REFERENCES h5_agent_runs(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  // MindSpace conversation packages: additive provenance tables for grouping
  // images, files, pages, and publications created during a chat session.
  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_conversation_packages (
      id VARCHAR(64) PRIMARY KEY,
      user_id CHAR(36) NOT NULL,
      session_id VARCHAR(128) NOT NULL,
      title VARCHAR(255) NULL,
      status ENUM('active', 'archived', 'deleted') NOT NULL DEFAULT 'active',
      storage_prefix VARCHAR(512) NULL,
      manifest_asset_id CHAR(36) NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      UNIQUE KEY uq_h5_conversation_package_session (user_id, session_id),
      KEY idx_h5_conversation_package_user_updated (user_id, updated_at),
      CONSTRAINT fk_h5_conversation_package_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE,
      CONSTRAINT fk_h5_conversation_package_manifest FOREIGN KEY (manifest_asset_id) REFERENCES h5_assets(id) ON DELETE SET NULL
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_conversation_artifacts (
      id VARCHAR(64) PRIMARY KEY,
      package_id VARCHAR(64) NOT NULL,
      asset_id CHAR(36) NULL,
      page_id CHAR(36) NULL,
      publication_id CHAR(36) NULL,
      agent_run_id CHAR(36) NULL,
      message_id VARCHAR(128) NULL,
      role VARCHAR(32) NULL,
      artifact_kind VARCHAR(64) NOT NULL,
      display_name VARCHAR(255) NULL,
      mime_type VARCHAR(128) NULL,
      size_bytes BIGINT NULL,
      storage_key VARCHAR(512) NULL,
      canonical_url VARCHAR(512) NULL,
      sort_order INT NOT NULL DEFAULT 0,
      created_at BIGINT NOT NULL,
      KEY idx_h5_conversation_artifact_package (package_id, sort_order, created_at),
      KEY idx_h5_conversation_artifact_asset (asset_id),
      KEY idx_h5_conversation_artifact_page (page_id),
      KEY idx_h5_conversation_artifact_publication (publication_id),
      KEY idx_h5_conversation_artifact_run (agent_run_id),
      KEY idx_h5_conversation_artifact_message (message_id),
      CONSTRAINT fk_h5_conversation_artifact_package FOREIGN KEY (package_id) REFERENCES h5_conversation_packages(id) ON DELETE CASCADE,
      CONSTRAINT fk_h5_conversation_artifact_asset FOREIGN KEY (asset_id) REFERENCES h5_assets(id) ON DELETE SET NULL,
      CONSTRAINT fk_h5_conversation_artifact_page FOREIGN KEY (page_id) REFERENCES h5_page_records(id) ON DELETE SET NULL,
      CONSTRAINT fk_h5_conversation_artifact_publication FOREIGN KEY (publication_id) REFERENCES h5_publish_records(id) ON DELETE SET NULL,
      CONSTRAINT fk_h5_conversation_artifact_run FOREIGN KEY (agent_run_id) REFERENCES h5_agent_runs(id) ON DELETE SET NULL
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  const uploadSessionColumns = [
    ['source_session_id', 'VARCHAR(128) NULL AFTER completed_asset_id'],
    ['source_message_id', 'VARCHAR(128) NULL AFTER source_session_id'],
  ];
  for (const [column, definition] of uploadSessionColumns) {
    if (!(await columnExists(pool, 'h5_upload_sessions', column))) {
      await pool.query(`ALTER TABLE h5_upload_sessions ADD COLUMN \`${column}\` ${definition}`);
    }
  }
  if (!(await indexExists(pool, 'h5_upload_sessions', 'idx_h5_upload_source_session'))) {
    await pool.query(
      `ALTER TABLE h5_upload_sessions
       ADD KEY idx_h5_upload_source_session (user_id, source_session_id)`,
    );
  }

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_user_memory_items (
      id CHAR(64) PRIMARY KEY,
      user_id CHAR(36) NOT NULL,
      label ENUM('preference', 'habit', 'interest', 'goal', 'fact', 'experience', 'knowledge') NOT NULL DEFAULT 'fact',
      memory_hash CHAR(64) NOT NULL,
      memory_text TEXT NOT NULL,
      evidence_message_id CHAR(64) NULL,
      source_session_id VARCHAR(128) NULL,
      confidence DECIMAL(4,3) NOT NULL DEFAULT 0.600,
      status ENUM('active', 'archived', 'deleted') NOT NULL DEFAULT 'active',
      raw_json JSON NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      UNIQUE KEY uq_h5_user_memory_hash (user_id, memory_hash),
      KEY idx_h5_user_memory_user_label (user_id, label, updated_at),
      KEY idx_h5_user_memory_session (source_session_id),
      CONSTRAINT fk_h5_user_memory_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE,
      CONSTRAINT fk_h5_user_memory_evidence FOREIGN KEY (evidence_message_id) REFERENCES h5_conversation_messages(id) ON DELETE SET NULL,
      CONSTRAINT fk_h5_user_memory_session FOREIGN KEY (source_session_id) REFERENCES h5_user_sessions(agent_session_id) ON DELETE SET NULL
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  const publishColumns = [
    ['user_confirmed_at', 'BIGINT NULL AFTER expires_at'],
    ['plaza_view_count', 'BIGINT NOT NULL DEFAULT 0'],
    ['plaza_like_count', 'BIGINT NOT NULL DEFAULT 0'],
  ];
  for (const [column, definition] of publishColumns) {
    if (!(await columnExists(pool, 'h5_publish_records', column))) {
      await pool.query(`ALTER TABLE h5_publish_records ADD COLUMN \`${column}\` ${definition}`);
    }
  }

  if (!(await indexExists(pool, 'h5_publish_records', 'idx_h5_publish_auto_private'))) {
    await pool.query(
      `ALTER TABLE h5_publish_records
       ADD KEY idx_h5_publish_auto_private (access_mode, status, user_confirmed_at, expires_at)`,
    );
  }

  const plazaUserColumns = [
    ['plaza_post_count', 'INT UNSIGNED NOT NULL DEFAULT 0'],
    ['plaza_follower_count', 'INT UNSIGNED NOT NULL DEFAULT 0'],
    ['plaza_following_count', 'INT UNSIGNED NOT NULL DEFAULT 0'],
    ['plaza_verified', 'TINYINT(1) NOT NULL DEFAULT 0'],
    ['plaza_post_banned', 'TINYINT(1) NOT NULL DEFAULT 0'],
    ['plaza_comment_banned', 'TINYINT(1) NOT NULL DEFAULT 0'],
    [
      'ops_role',
      "ENUM('none','reviewer','editor','ops_admin') NOT NULL DEFAULT 'none'",
    ],
  ];
  for (const [column, definition] of plazaUserColumns) {
    if (!(await columnExists(pool, 'h5_users', column))) {
      await pool.query(`ALTER TABLE h5_users ADD COLUMN \`${column}\` ${definition}`);
    }
  }

  if (!(await columnExists(pool, 'h5_users', 'signup_source'))) {
    await pool.query(
      `ALTER TABLE h5_users ADD COLUMN signup_source VARCHAR(32) NULL DEFAULT 'password' AFTER workspace_root`,
    );
  }

  if (!(await columnExists(pool, 'h5_users', 'low_balance_gift_eligible'))) {
    await pool.query(
      `ALTER TABLE h5_users ADD COLUMN low_balance_gift_eligible TINYINT(1) NOT NULL DEFAULT 0 AFTER signup_source`,
    );
  }

  if (!(await columnExists(pool, 'h5_users', 'low_balance_gift_granted_at'))) {
    await pool.query(
      `ALTER TABLE h5_users ADD COLUMN low_balance_gift_granted_at BIGINT NULL AFTER low_balance_gift_eligible`,
    );
  }

  if (!(await columnExists(pool, 'h5_user_sessions', 'goosed_node'))) {
    await pool.query(
      `ALTER TABLE h5_user_sessions ADD COLUMN goosed_node TINYINT UNSIGNED NOT NULL DEFAULT 0`,
    );
  }

  // goosed_target stores the resolved upstream URL a session is pinned to, so
  // routing survives reordering or resizing the goosed target list. The legacy
  // integer goosed_node is kept as a fallback for rows written before this column
  // existed (and for older bundles that still read it).
  if (!(await columnExists(pool, 'h5_user_sessions', 'goosed_target'))) {
    await pool.query(
      `ALTER TABLE h5_user_sessions ADD COLUMN goosed_target VARCHAR(255) NULL AFTER goosed_node`,
    );
  }

  if (!(await columnExists(pool, 'h5_user_sessions', 'origin'))) {
    await pool.query(
      `ALTER TABLE h5_user_sessions ADD COLUMN origin ENUM('h5', 'wechat') NOT NULL DEFAULT 'h5' AFTER user_id`,
    );
  }

  if (!(await columnExists(pool, 'h5_session_snapshots', 'display_title'))) {
    await pool.query(
      `ALTER TABLE h5_session_snapshots ADD COLUMN display_title VARCHAR(512) NOT NULL DEFAULT '' AFTER name`,
    );
  }

  const oauthStateColumns = [
    ['intent', "VARCHAR(16) NOT NULL DEFAULT 'login' AFTER utm_campaign"],
    ['bind_user_id', 'CHAR(36) NULL AFTER intent'],
    ['auth_mode', "VARCHAR(8) NOT NULL DEFAULT 'mp' AFTER bind_user_id"],
    ['status', "VARCHAR(16) NOT NULL DEFAULT 'pending' AFTER auth_mode"],
    ['result_kind', 'VARCHAR(32) NULL AFTER status'],
    ['result_token', 'VARCHAR(512) NULL AFTER result_kind'],
    ['result_message', 'VARCHAR(255) NULL AFTER result_token'],
  ];
  for (const [column, definition] of oauthStateColumns) {
    if (!(await columnExists(pool, 'h5_wechat_oauth_states', column))) {
      await pool.query(
        `ALTER TABLE h5_wechat_oauth_states ADD COLUMN \`${column}\` ${definition}`,
      );
    }
  }

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_wechat_pending_binds (
      token VARCHAR(64) PRIMARY KEY,
      app_id VARCHAR(32) NOT NULL,
      openid VARCHAR(64) NOT NULL,
      unionid VARCHAR(64) NULL,
      nickname VARCHAR(128) NULL,
      avatar_url VARCHAR(512) NULL,
      return_to VARCHAR(512) NULL,
      utm_source VARCHAR(64) NULL,
      utm_medium VARCHAR(64) NULL,
      utm_campaign VARCHAR(64) NULL,
      expires_at BIGINT NOT NULL,
      created_at BIGINT NOT NULL,
      KEY idx_wechat_pending_expires (expires_at)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_wechat_mp_messages (
      app_id VARCHAR(32) NOT NULL,
      openid VARCHAR(64) NOT NULL,
      msg_id VARCHAR(128) NOT NULL,
      status ENUM('processing', 'done', 'failed') NOT NULL DEFAULT 'processing',
      agent_session_id VARCHAR(128) NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      PRIMARY KEY (app_id, openid, msg_id),
      KEY idx_wechat_mp_messages_updated (updated_at),
      KEY idx_wechat_mp_messages_session (agent_session_id)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_wechat_mp_message_details (
      id CHAR(36) PRIMARY KEY,
      app_id VARCHAR(32) NOT NULL,
      openid VARCHAR(64) NOT NULL,
      user_id CHAR(36) NULL,
      msg_id VARCHAR(128) NULL,
      msg_type VARCHAR(32) NOT NULL,
      display_text TEXT NULL,
      agent_text TEXT NULL,
      media_id VARCHAR(256) NULL,
      media_url TEXT NULL,
      media_public_url TEXT NULL,
      media_format VARCHAR(64) NULL,
      location_lat DECIMAL(10,7) NULL,
      location_lng DECIMAL(10,7) NULL,
      location_label VARCHAR(255) NULL,
      link_url TEXT NULL,
      link_title VARCHAR(255) NULL,
      raw_xml_hash CHAR(40) NULL,
      raw_json JSON NULL,
      created_at BIGINT NOT NULL,
      KEY idx_wechat_mp_message_details_user (app_id, openid, created_at),
      KEY idx_wechat_mp_message_details_msg (msg_id),
      KEY idx_wechat_mp_message_details_type (msg_type, created_at),
      CONSTRAINT fk_wechat_mp_message_details_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE SET NULL
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_schedule_items (
      id CHAR(36) PRIMARY KEY,
      user_id CHAR(36) NOT NULL,
      kind ENUM('task', 'event') NOT NULL,
      title VARCHAR(255) NOT NULL,
      description TEXT NULL,
      status ENUM('active', 'completed', 'cancelled', 'deleted') NOT NULL DEFAULT 'active',
      start_at BIGINT NULL,
      end_at BIGINT NULL,
      due_at BIGINT NULL,
      all_day TINYINT(1) NOT NULL DEFAULT 0,
      timezone VARCHAR(64) NOT NULL DEFAULT 'Asia/Shanghai',
      location VARCHAR(255) NULL,
      source_channel ENUM('h5', 'wechat', 'agent', 'api') NOT NULL DEFAULT 'agent',
      source_session_id VARCHAR(128) NULL,
      source_message_id VARCHAR(128) NULL,
      source_text TEXT NULL,
      metadata_json JSON NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      deleted_at BIGINT NULL,
      KEY idx_schedule_user_time (user_id, status, start_at, due_at),
      KEY idx_schedule_user_updated (user_id, updated_at),
      CONSTRAINT fk_schedule_item_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_schedule_reminders (
      id CHAR(36) PRIMARY KEY,
      user_id CHAR(36) NOT NULL,
      item_id CHAR(36) NOT NULL,
      remind_at BIGINT NOT NULL,
      offset_minutes INT NULL,
      channel ENUM('wechat', 'in_app') NOT NULL DEFAULT 'wechat',
      status ENUM('pending', 'locked', 'sent', 'failed', 'cancelled') NOT NULL DEFAULT 'pending',
      attempts INT NOT NULL DEFAULT 0,
      last_error VARCHAR(500) NULL,
      locked_until BIGINT NULL,
      sent_at BIGINT NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      UNIQUE KEY uq_schedule_item_remind_at (item_id, remind_at, channel),
      KEY idx_reminder_due (status, remind_at),
      KEY idx_reminder_user (user_id, status, remind_at),
      CONSTRAINT fk_schedule_reminder_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE,
      CONSTRAINT fk_schedule_reminder_item FOREIGN KEY (item_id) REFERENCES h5_schedule_items(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_schedule_digest_subscriptions (
      id CHAR(36) PRIMARY KEY,
      user_id CHAR(36) NOT NULL,
      digest_type ENUM('todo_day') NOT NULL DEFAULT 'todo_day',
      hour TINYINT UNSIGNED NOT NULL,
      minute TINYINT UNSIGNED NOT NULL DEFAULT 0,
      timezone VARCHAR(64) NOT NULL DEFAULT 'Asia/Shanghai',
      channel ENUM('wechat', 'in_app') NOT NULL DEFAULT 'wechat',
      status ENUM('active', 'locked', 'failed', 'cancelled') NOT NULL DEFAULT 'active',
      next_run_at BIGINT NOT NULL,
      last_run_at BIGINT NULL,
      attempts INT NOT NULL DEFAULT 0,
      locked_until BIGINT NULL,
      last_error VARCHAR(500) NULL,
      source_channel ENUM('h5', 'wechat', 'agent', 'api') NOT NULL DEFAULT 'agent',
      source_session_id VARCHAR(128) NULL,
      source_message_id VARCHAR(128) NULL,
      source_text TEXT NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      UNIQUE KEY uq_schedule_digest_user_type_channel (user_id, digest_type, channel),
      KEY idx_schedule_digest_due (status, next_run_at),
      CONSTRAINT fk_schedule_digest_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_balance_alert_subscriptions (
      id CHAR(36) PRIMARY KEY,
      user_id CHAR(36) NOT NULL,
      threshold_cents BIGINT NOT NULL,
      channel ENUM('wechat', 'in_app') NOT NULL DEFAULT 'wechat',
      status ENUM('active', 'locked', 'failed', 'cancelled') NOT NULL DEFAULT 'active',
      next_run_at BIGINT NOT NULL,
      last_run_at BIGINT NULL,
      last_notified_balance_cents BIGINT NULL,
      attempts INT NOT NULL DEFAULT 0,
      locked_until BIGINT NULL,
      last_error VARCHAR(500) NULL,
      source_channel ENUM('h5', 'wechat', 'agent', 'api') NOT NULL DEFAULT 'agent',
      source_session_id VARCHAR(128) NULL,
      source_message_id VARCHAR(128) NULL,
      source_text TEXT NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      UNIQUE KEY uq_balance_alert_user_channel (user_id, channel),
      KEY idx_balance_alert_due (status, next_run_at),
      KEY idx_balance_alert_user (user_id, status, next_run_at),
      CONSTRAINT fk_balance_alert_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_schedule_delivery_logs (
      id CHAR(36) PRIMARY KEY,
      reminder_id CHAR(36) NULL,
      subscription_id CHAR(36) NULL,
      user_id CHAR(36) NOT NULL,
      channel ENUM('wechat', 'in_app') NOT NULL,
      status ENUM('success', 'failed') NOT NULL,
      provider_message_id VARCHAR(128) NULL,
      error_code VARCHAR(64) NULL,
      error_message VARCHAR(500) NULL,
      created_at BIGINT NOT NULL,
      KEY idx_delivery_reminder (reminder_id, created_at),
      KEY idx_delivery_subscription (subscription_id, created_at),
      KEY idx_delivery_user (user_id, created_at),
      CONSTRAINT fk_schedule_delivery_reminder FOREIGN KEY (reminder_id) REFERENCES h5_schedule_reminders(id) ON DELETE CASCADE,
      CONSTRAINT fk_schedule_delivery_subscription FOREIGN KEY (subscription_id) REFERENCES h5_schedule_digest_subscriptions(id) ON DELETE CASCADE,
      CONSTRAINT fk_schedule_delivery_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_user_notifications (
      id CHAR(36) PRIMARY KEY,
      user_id CHAR(36) NOT NULL,
      channel ENUM('web', 'wechat') NOT NULL DEFAULT 'web',
      notification_type VARCHAR(64) NOT NULL,
      title VARCHAR(255) NOT NULL,
      body TEXT NOT NULL,
      data_json JSON NULL,
      status ENUM('unread', 'read') NOT NULL DEFAULT 'unread',
      read_at BIGINT NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      KEY idx_user_notifications_user_status_created (user_id, status, created_at),
      KEY idx_user_notifications_user_created (user_id, created_at),
      CONSTRAINT fk_user_notifications_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  await pool.query(
    `ALTER TABLE h5_payment_orders
     MODIFY pay_mode ENUM('native', 'h5', 'jsapi') NOT NULL DEFAULT 'native'`,
  );

  // Session snapshot cache table (additive, no existing-data impact).
  // Rollback: DROP TABLE h5_session_snapshots — safe, no dependants.
  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_session_snapshots (
      agent_session_id VARCHAR(128) NOT NULL PRIMARY KEY,
      user_id CHAR(36) NOT NULL,
      name VARCHAR(512) NOT NULL DEFAULT '',
      display_title VARCHAR(512) NOT NULL DEFAULT '',
      working_dir VARCHAR(1024) NOT NULL DEFAULT '',
      created_at_str VARCHAR(64) NOT NULL DEFAULT '',
      updated_at_str VARCHAR(64) NOT NULL DEFAULT '',
      user_set_name TINYINT(1) NOT NULL DEFAULT 0,
      recipe_json TEXT NULL,
      synced_msg_count INT NOT NULL DEFAULT 0,
      source_updated_at VARCHAR(64) NOT NULL DEFAULT '',
      messages_json LONGTEXT NOT NULL,
      synced_at BIGINT NOT NULL,
      KEY idx_h5_snapshots_user (user_id),
      KEY idx_h5_snapshots_synced (synced_at),
      CONSTRAINT fk_h5_snapshot_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  // Session stream replay events (Patch 4c). Rollback: DROP TABLE h5_session_stream_events.
  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_session_stream_events (
      id CHAR(36) PRIMARY KEY,
      user_id CHAR(36) NOT NULL,
      agent_session_id VARCHAR(128) NOT NULL,
      event_type VARCHAR(64) NOT NULL,
      payload_json LONGTEXT NOT NULL,
      upstream_event_id VARCHAR(255) NULL,
      created_at BIGINT NOT NULL,
      KEY idx_h5_session_stream_session (agent_session_id, created_at, id),
      KEY idx_h5_session_stream_user (user_id, created_at),
      CONSTRAINT fk_h5_session_stream_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  // Subscription plan table for tiered monthly billing.
  // Rollback: DROP TABLE h5_subscriptions — safe, no dependants.
  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_subscriptions (
      id CHAR(36) PRIMARY KEY,
      user_id CHAR(36) NOT NULL,
      plan_type VARCHAR(32) NOT NULL DEFAULT 'free',
      status ENUM('active', 'expired', 'cancelled') NOT NULL DEFAULT 'active',
      period_tokens_limit BIGINT NOT NULL DEFAULT 0,
      period_tokens_used BIGINT NOT NULL DEFAULT 0,
      period_start BIGINT NOT NULL,
      period_end BIGINT NOT NULL,
      expires_at BIGINT NOT NULL,
      overage_rate DECIMAL(4,2) NOT NULL DEFAULT 1.00,
      operator_id CHAR(36) NULL,
      note VARCHAR(512) NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      KEY idx_h5_sub_user_status (user_id, status),
      KEY idx_h5_sub_expires (expires_at, status),
      CONSTRAINT fk_h5_sub_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);

  // Back-fill free subscriptions for existing users who pre-date the subscription system.
  // Idempotent: only inserts where no active subscription exists.
  await pool.query(`
    INSERT INTO h5_subscriptions
      (id, user_id, plan_type, status,
       period_tokens_limit, period_tokens_used,
       period_start, period_end, expires_at, overage_rate,
       operator_id, note, created_at, updated_at)
    SELECT
      UUID(),
      u.id,
      'free',
      'active',
      150000, 0,
      UNIX_TIMESTAMP() * 1000,
      (UNIX_TIMESTAMP() + 30 * 86400) * 1000,
      (UNIX_TIMESTAMP() + 30 * 86400) * 1000,
      1.00,
      NULL,
      '系统迁移补建免费套餐',
      UNIX_TIMESTAMP() * 1000,
      UNIX_TIMESTAMP() * 1000
    FROM h5_users u
    WHERE NOT EXISTS (
      SELECT 1 FROM h5_subscriptions s
      WHERE s.user_id = u.id AND s.status = 'active'
    )
  `);

  if (!(await columnExists(pool, 'h5_page_records', 'workspace_relative_path'))) {
    await pool.query(
      `ALTER TABLE h5_page_records
       ADD COLUMN workspace_relative_path VARCHAR(512) NULL AFTER draft_content_ref`,
    );
  }
  if (!(await indexExists(pool, 'h5_page_records', 'idx_h5_pages_workspace_path'))) {
    await pool.query(
      `ALTER TABLE h5_page_records
       ADD KEY idx_h5_pages_workspace_path (user_id, workspace_relative_path, updated_at)`,
    );
  }
  if (!(await columnExists(pool, 'h5_assets', 'workspace_relative_path'))) {
    await pool.query(
      `ALTER TABLE h5_assets
       ADD COLUMN workspace_relative_path VARCHAR(512) NULL AFTER display_name`,
    );
  }
  if (!(await indexExists(pool, 'h5_assets', 'idx_h5_assets_workspace_path'))) {
    await pool.query(
      `ALTER TABLE h5_assets
       ADD KEY idx_h5_assets_workspace_path (user_id, workspace_relative_path, updated_at)`,
    );
  }
  await pool.query(
    `UPDATE h5_page_records p
     JOIN h5_page_versions pv ON pv.id = p.current_version_id
     SET p.workspace_relative_path = JSON_UNQUOTE(JSON_EXTRACT(pv.source_snapshot_json, '$.relative_path'))
     WHERE p.workspace_relative_path IS NULL
       AND JSON_EXTRACT(pv.source_snapshot_json, '$.relative_path') IS NOT NULL
       AND JSON_UNQUOTE(JSON_EXTRACT(pv.source_snapshot_json, '$.relative_path')) <> ''`,
  );
  await pool.query(
    `UPDATE h5_assets a
     JOIN h5_space_categories c ON c.id = a.category_id AND c.user_id = a.user_id
     SET a.workspace_relative_path = CASE
       WHEN a.original_filename LIKE 'public/%' OR a.original_filename LIKE 'oa/%'
         THEN a.original_filename
       WHEN c.category_code IN ('public', 'oa')
         THEN CONCAT(c.category_code, '/', a.original_filename)
       ELSE NULL
     END
     WHERE a.workspace_relative_path IS NULL
       AND a.status <> 'deleted'
       AND c.category_code IN ('public', 'oa')`,
  );

  await pool.query(`
    CREATE TABLE IF NOT EXISTS h5_user_feedback (
      id CHAR(36) PRIMARY KEY,
      user_id CHAR(36) NOT NULL,
      type ENUM('bug', 'feature', 'other') NOT NULL,
      title VARCHAR(120) NOT NULL,
      description TEXT NOT NULL,
      contact VARCHAR(120) NULL,
      status ENUM('pending', 'reviewing', 'resolved', 'closed') NOT NULL DEFAULT 'pending',
      images_json JSON NULL,
      context_json JSON NULL,
      created_at BIGINT NOT NULL,
      updated_at BIGINT NOT NULL,
      KEY idx_h5_user_feedback_user_created (user_id, created_at),
      KEY idx_h5_user_feedback_status_created (status, created_at),
      CONSTRAINT fk_h5_user_feedback_user FOREIGN KEY (user_id) REFERENCES h5_users(id) ON DELETE CASCADE
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  `);
}

export async function initSchema(pool) {
  const schemaPath = path.join(__dirname, 'schema.sql');
  const sql = fs
    .readFileSync(schemaPath, 'utf8')
    .split('\n')
    .filter((line) => !line.trim().startsWith('--'))
    .join('\n');
  for (const statement of sql.split(';')) {
    const trimmed = statement.trim();
    if (trimmed) {
      await pool.query(trimmed);
    }
  }
  await migrateSchema(pool);
  await ensureDefaultSpaces(pool, {
    quotaBytes: Number(process.env.MINDSPACE_FREE_QUOTA_BYTES ?? 5 * 1024 * 1024),
  });
}