import { Injectable, Logger, OnModuleInit } from '@nestjs/common';
import { SchedulerRegistry } from '@nestjs/schedule';
import { CronJob } from 'cron';
import * as path from 'path';
import { SyncStatus } from '@pac/types';
import { PrismaService } from '../prisma/prisma.service';
import {
  ColdImportService,
  SyncAlreadyRunningError,
} from '../modules/sync/cold-import/cold-import.service';

/**
 * 本进程启动时刻(模块加载即固定)。
 * 用作遗留锁判据:任何 startedAt 早于此刻的 running sync_log,其进程必然早于本进程启动 ——
 * 而 sync 进程 = 本 service 进程(cron 内跑)或一次性 CLI(跑完即退)—— 都不可能还活着。
 */
const PROCESS_STARTED_AT = new Date();
import { OrgTreeService } from '../modules/auth/org-tree';
import { PersonaService } from '../modules/persona/persona.service';
import { PlanEngineService } from '../modules/plan/engine/plan-engine.service';
import { schedulerDisabled } from './scheduler-switch';

/**
 * SyncIncrementalSchedulerService — DW 直连增量自动跑(**每宿主独立 cron**)
 *
 * 每天在各 host 自己的 DW 刷新窗口后触发:
 *   1. ColdImportService.importDirectory({ incremental: true })(游标增量 + stub 补齐)
 *   2. Persona 重算(本次 sync 写入的 distinct patient_id)
 *   3. org-tree 预热 + Plan 重算(per tenant)
 *
 * **多 host 调度**(启动时动态注册,取代单一全局 @Cron):
 *   - 发现 data/<host>/manifest.yaml 中 auto_sync=true 的 host;
 *   - 每个 host 用**自己 manifest 的 incremental_cron + timezone**注册独立 CronJob
 *     —— 上游 DW 刷新窗口因 host 而异,各自在就绪后跑,不再挤在一个全局时刻;
 *   - manifest 没配 cron → 回退全局 PAC_INCREMENTAL_CRON(按 host manifest 时区解释)。
 *   - PAC_INCREMENTAL_HOSTS env(逗号分隔)仍作显式 override 逃生口(用全局 cron)。
 *
 * 跑失败:cursor 不前进 → 下次自动 catchup;log ERROR 不抛。
 */
@Injectable()
export class SyncIncrementalSchedulerService implements OnModuleInit {
  private readonly logger = new Logger(SyncIncrementalSchedulerService.name);

  constructor(
    private readonly prisma: PrismaService,
    private readonly coldImport: ColdImportService,
    private readonly orgTree: OrgTreeService,
    private readonly persona: PersonaService,
    private readonly planEngine: PlanEngineService,
    private readonly scheduler: SchedulerRegistry,
  ) {}

  async onModuleInit(): Promise<void> {
    // ⭐ 写侧总闸:PAC_SCHEDULER_DISABLED=1 → 本实例不跑任何**会改数据**的定时任务。
    //   用于「新旧实例并存」的窗口(服务器迁移 / 蓝绿 / 备用实例):新实例要能起来接流量、
    //   验证配置,但**绝不能**同时对 DW 做增量摄入 + persona/plan 重算 —— 两边各拉各的,
    //   会让待迁移的数据快照持续偏移;若两边连同一个库,还会互抢 sync_logs 的
    //   partial UNIQUE(host_id) WHERE status='running' 锁、互相回收对方的"僵尸锁"。
    //
    //   ⚠️ 为什么必须是代码开关:jvs-dw 的 manifest 写死 auto_sync=true,cron 也写在
    //   manifest(优先级高于 env),`PAC_INCREMENTAL_HOSTS=` 留空只会 fallback 到自动发现 ——
    //   **靠 env 关不掉**。改 manifest 又会被 deploy 的 git pull 覆盖。
    //
    //   ⚠️ 连僵尸锁回收也一并跳过:回收判据是"startedAt 早于本进程启动",共库场景下
    //   新实例会把老实例**正在跑**的锁误当僵尸回收掉。禁用态就该完全不碰 sync_logs。
    if (schedulerDisabled()) {
      this.logger.warn(
        'sync-incremental: ⏸ PAC_SCHEDULER_DISABLED=1 —— 不注册 cron、不回收僵尸锁(迁移/备用实例)',
      );
      return;
    }

    // 启动即回收上一个进程留下的僵尸锁(部署/崩溃重启时,running 行来不及标终态,
    // 会永久占住 sync_logs 的 partial UNIQUE(host_id) WHERE status='running',
    // 导致之后每次 cron 增量都被并发锁 skip → 游标不动 → DW 滞后告警)。
    await this.reapStaleRunningLocks();

    const globalCron = process.env.PAC_INCREMENTAL_CRON;
    const dataDir = this.dataDir();
    // env override(逗号分隔)优先;否则自动发现 auto_sync=true 的 host
    const envHosts = (process.env.PAC_INCREMENTAL_HOSTS ?? '')
      .split(',')
      .map((h) => h.trim())
      .filter(Boolean);
    const hosts = envHosts.length > 0 ? envHosts : this.coldImport.discoverAutoSyncHostDirs(dataDir);

    if (hosts.length === 0) {
      this.logger.log('sync-incremental: 无 auto_sync host,不注册 cron');
      return;
    }

    for (const host of hosts) {
      const ops = this.coldImport.getHostOpsConfig(host);
      // 每宿主 cron:manifest.incremental_cron 优先,否则全局 env;都没有 → 不注册(该 host 不自动同步)
      const cronExpr = ops.incrementalCron ?? globalCron;
      if (!cronExpr) {
        this.logger.warn(
          `sync-incremental: host=${host} 无 cron(manifest.incremental_cron 与 PAC_INCREMENTAL_CRON 都未设),跳过注册`,
        );
        continue;
      }
      const timeZone = ops.timezone ?? 'Asia/Shanghai';
      const jobName = `sync-incremental:${host}`;
      const job = new CronJob(
        cronExpr,
        () => {
          void this.runHostSafe(host);
        },
        null,
        false,
        timeZone,
      );
      this.scheduler.addCronJob(jobName, job as unknown as Parameters<SchedulerRegistry['addCronJob']>[1]);
      job.start();
      this.logger.log(
        `sync-incremental: 注册 host=${host} cron='${cronExpr}' tz=${timeZone}` +
          `${ops.incrementalCron ? '' : '(全局默认)'}`,
      );
    }
  }

  private dataDir(): string {
    return process.env.PAC_INCREMENTAL_DATA_DIR ?? path.resolve(__dirname, '../../data');
  }

  /**
   * 回收僵尸同步锁 —— 把 startedAt 早于本进程启动的 running sync_log 标 failed。
   *
   * 为什么这样判据安全:sync 只在两处跑 —— 本 service 进程的 cron 回调,或一次性 CLI
   * (`sync-incremental.cli` / `cold-import.cli`,跑完即退)。两者的进程都不可能比本进程
   * 启动得更早还活着。所以 `startedAt < PROCESS_STARTED_AT` 的 running 行 = 上一个已死进程的残留。
   * 反过来,本进程启动后新建的 running 行(startedAt >= PROCESS_STARTED_AT)绝不碰 —— 那可能是
   * 正在跑的真锁(例如运维手动触发的 CLI 与本进程并存),误清会把在跑的同步" orphan"掉。
   *
   * 幂等:被标 failed 只是让并发锁释放;数据侧不受影响(游标没推进,下次增量靠 48h 回看窗补齐)。
   *
   * @param processStartedAt 回收判据的时间界(默认本进程启动时刻)。显式传入是为了让判据可测 ——
   *   测试不必靠真实时钟凑时间差(高负载下事件循环被拖慢会越过阈值边界,产生间歇性假失败)。
   */
  private async reapStaleRunningLocks(processStartedAt: Date = PROCESS_STARTED_AT): Promise<void> {
    try {
      const stale = await this.prisma.syncLog.findMany({
        where: { status: SyncStatus.RUNNING, startedAt: { lt: processStartedAt } },
        select: { id: true, hostId: true, startedAt: true, triggeredBy: true },
      });
      if (stale.length === 0) return;

      const { count } = await this.prisma.syncLog.updateMany({
        where: { status: SyncStatus.RUNNING, startedAt: { lt: processStartedAt } },
        data: {
          status: SyncStatus.FAILED,
          endedAt: new Date(),
          errorMessage: '进程重启前未完成,启动时回收僵尸锁(reapStaleRunningLocks)',
        },
      });
      for (const s of stale) {
        this.logger.warn(
          `sync-incremental: 回收僵尸锁 host=${s.hostId} started=${s.startedAt.toISOString()} ` +
            `trigger=${s.triggeredBy}`,
        );
      }
      this.logger.log(`sync-incremental: 启动回收 ${count} 个僵尸同步锁`);
    } catch (err) {
      // 回收失败不阻断启动(cron 仍注册);最坏退回人工清理,不比现状糟
      this.logger.error(`sync-incremental: 僵尸锁回收失败: ${(err as Error).message}`);
    }
  }

  /// 单 host 跑一轮(cron 回调用,吞异常不影响该 host 下次 / 别的 host)
  private async runHostSafe(host: string): Promise<void> {
    try {
      await this.runOne(path.join(this.dataDir(), host));
    } catch (err) {
      if (err instanceof SyncAlreadyRunningError) {
        this.logger.warn(`sync-incremental: host=${host} skip(并发锁拦截):${err.message}`);
      } else {
        this.logger.error(`sync-incremental: host=${host} failed: ${(err as Error).message}`);
      }
    }
  }

  /// 跑单个 host 的完整链路(可手动调,debug 用)
  async runOne(dir: string): Promise<void> {
    const started = Date.now();
    const r = await this.coldImport.importDirectory(dir, { incremental: true });
    this.logger.log(
      `sync-incremental: cold-import done host=${r.hostName} ` +
        `patients=${r.totals.patientsUpserted} txns=${r.totals.transactionsWritten} ` +
        `facts(created=${r.totals.factsCreated} superseded=${r.totals.factsSuperseded})`,
    );

    // ─── 空轮短路(2h 提频配套)──────
    // DW 每日一批 → 日内轮询多为空轮;没摄到任何数据时 persona/org-tree/plan 全跳,
    // 空轮成本收敛到"每表一条 cursor 探查"(秒级)。条件保守:患者主档或事务任一有
    // 变化都算有数据(patient-only 变化不产事务,但会进画像,不能跳)。
    if (r.totals.transactionsWritten === 0 && r.totals.patientsUpserted === 0) {
      this.logger.log(
        `sync-incremental: host=${r.hostName} 空轮(0 新数据),跳过 persona/org-tree/plan ` +
          `(elapsed=${Date.now() - started}ms)`,
      );
      return;
    }

    // ─── Persona recompute(affected patient only)──────
    if (r.syncLogId) {
      const syncLog = await this.prisma.syncLog.findUnique({ where: { id: r.syncLogId } });
      if (syncLog) {
        const affected = await this.prisma.patientTransaction.findMany({
          where: { hostId: r.hostId, createdAt: { gte: syncLog.startedAt } },
          select: { patientId: true },
          distinct: ['patientId'],
        });
        const pids = affected.map((p) => p.patientId).filter((p): p is string => !!p);
        let ok = 0;
        let failed = 0;
        // 串行:persona 重算是 **Node-CPU-bound**(16 特征 JS 计算占主),实测 conc=8 不提速、
        //   还因竞争略慢(本地 300 患者 30s→32s)。异步并发只能重叠 I/O,重叠不了单线程 CPU。
        //   真要提速需 profile 特征计算或多进程/worker,不在此处做。
        for (const pid of pids) {
          try {
            await this.persona.recompute({
              patientId: pid,
              source: `incremental-sync:${r.runId}`,
            });
            ok++;
          } catch (err) {
            failed++;
            this.logger.warn(`persona patient=${pid}: ${(err as Error).message}`);
          }
        }
        this.logger.log(`persona recompute: success=${ok} failed=${failed} (${pids.length} affected)`);
      }
    }

    const tenants = await this.prisma.patient.findMany({
      where: { hostId: r.hostId },
      distinct: ['tenantId'],
      select: { tenantId: true },
    });

    // ─── org-tree 主动预热(消除"当天第一个非 admin 登录用户撞冷派生"的尾延迟)──────
    // 本次同步刚写完新数据,强制重派生一次(forceRefresh),后续 24h TTL 内任何用户登录
    // 都直接命中缓存 —— 冷派生的代价(生产 379 万行,实测约 5s)由本 cron(后台)吞掉,
    // 不再由真实用户请求承担。单个 tenant 预热失败不影响其余 tenant / 后续 sync 步骤。
    for (const t of tenants) {
      try {
        await this.orgTree.getTree(r.hostId, t.tenantId, true);
      } catch (err) {
        this.logger.warn(
          `org-tree 预热失败 host=${r.hostId} tenant=${t.tenantId}: ${(err as Error).message}`,
        );
      }
    }

    // ─── Plan recompute(per tenant)──────
    let plansCreated = 0;
    for (const t of tenants) {
      const r2 = await this.planEngine.runAllForHost({
        hostId: r.hostId,
        tenantId: t.tenantId,
        now: new Date(),
      });
      plansCreated += r2.plansCreated;
    }
    this.logger.log(
      `plan recompute: ${tenants.length} tenants, plansCreated=${plansCreated} ` +
        `(total elapsed=${Date.now() - started}ms)`,
    );
  }
}
