import { Kysely, PostgresDialect, sql } from "kysely";
import pg from "pg";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { type DB, listenCacheInvalidation, MetadataCache, migrateToLatest, PgCache } from "../src/index.ts";

// 统一元数据缓存(0086):两级读写、跨进程失效、过期先用旧值、同请求合并、失败不缓存、pinned 不被淘汰、命中统计。
const url = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
const skip = process.env.UMT_SKIP_DB_TESTS === "1";
const schema = `umt_mcache_${process.pid}_${Date.now().toString(36)}`;

const until = async (cond: () => boolean | Promise<boolean>, ms = 5000) => {
  const end = Date.now() + ms;
  while (!(await cond())) {
    if (Date.now() > end) throw new Error("等待超时");
    await new Promise((r) => setTimeout(r, 20));
  }
};

describe.skipIf(skip)("MetadataCache(真实 PG)", () => {
  let admin: Kysely<unknown>;
  let db: Kysely<DB>;
  /** 另一个「进程」:独立连接池 */
  let db2: Kysely<DB>;

  const pool = () =>
    new Kysely<DB>({
      dialect: new PostgresDialect({
        pool: new pg.Pool({ connectionString: url, max: 4, options: `-c search_path=${schema},public` }),
      }),
    });

  beforeAll(async () => {
    admin = new Kysely({ dialect: new PostgresDialect({ pool: new pg.Pool({ connectionString: url, max: 1 }) }) });
    await sql.raw(`create schema ${schema}`).execute(admin);
    db = pool();
    db2 = pool();
    await migrateToLatest(db, { migrationTableSchema: schema });
  });

  afterAll(async () => {
    await db?.destroy();
    await db2?.destroy();
    await sql.raw(`drop schema if exists ${schema} cascade`).execute(admin);
    await admin?.destroy();
  });

  it("读 L1 → L2 → 源,写两级;另一进程从 L2 命中;统计各级命中", async () => {
    const a = new MetadataCache(new PgCache(db), { l1: true });
    const b = new MetadataCache(new PgCache(db2), { l1: true });
    let n = 0;
    const load = async () => ({ v: ++n });
    const p = { ttl: 600 };
    expect(await a.fetch("r1", "k", p, load)).toEqual({ v: 1 });
    expect(await a.fetch("r1", "k", p, load)).toEqual({ v: 1 });
    expect(await b.fetch("r1", "k", p, load)).toEqual({ v: 1 });
    expect(await b.fetch("r1", "k", p, load)).toEqual({ v: 1 });
    expect(n).toBe(1);
    expect(a.pendingStats().get("r1")).toMatchObject({ misses: 1, l1Hits: 1, l2Hits: 0 });
    expect(b.pendingStats().get("r1")).toMatchObject({ misses: 0, l2Hits: 1, l1Hits: 1 });
    // L1 返回副本:调用方改了不污染缓存
    const got = await a.fetch<{ v: number }>("r1", "k", p, load);
    got.v = 99;
    expect(await a.fetch("r1", "k", p, load)).toEqual({ v: 1 });
    // 有宽限期:fresh_until = now + ttl,expires_at 再加 ttl 的 50%
    const r = await sql<{ g: number }>`select extract(epoch from expires_at - fresh_until)::int as g
      from kv_cache where region = 'r1' and key = 'k'`.execute(db);
    expect(r.rows[0]!.g).toBe(300);
    // 统计写进 cache_stat 后清零
    expect(await a.flushStats()).toBe(1);
    await b.flushStats();
    expect(a.pendingStats().size).toBe(0);
    const s = await sql<{ l1: number; l2: number; m: number }>`select l1_hits::int as l1, l2_hits::int as l2,
      misses::int as m from cache_stat where region = 'r1'`.execute(db);
    expect(s.rows[0]).toEqual({ l1: 4, l2: 1, m: 1 });
  });

  it("跨进程失效:删除 / 清空通知其它进程丢 L1;监听断开时 L1 停用,重连后清空再启用", async () => {
    const a = new MetadataCache(new PgCache(db));
    const b = new MetadataCache(new PgCache(db2));
    const la = await listenCacheInvalidation(a, { connectionString: url });
    const lb = await listenCacheInvalidation(b, { connectionString: url });
    try {
      expect(a.l1Enabled && b.l1Enabled).toBe(true);
      const p = { ttl: 600 };
      await a.fetch("inv", "x", p, async () => 1);
      await a.fetch("inv", "y", p, async () => 2);
      await b.fetch("inv", "x", p, async () => 0);
      await b.fetch("inv", "y", p, async () => 0);
      expect(b.l1Size("inv")).toBe(2);
      // A 删 x:B 的 L1 只丢 x
      await a.delete("inv", "x");
      await until(() => b.l1Size("inv") === 1);
      expect(await b.fetch("inv", "x", p, async () => 10)).toBe(10);
      // A 清空整个 region
      await a.clear("inv");
      await until(() => b.l1Size("inv") === 0);
      expect(await b.fetch("inv", "y", p, async () => 20)).toBe(20);
      // 手动刷新(fresh)写入后也通知
      await a.fetch("inv", "y", { ...p, fresh: true }, async () => 21);
      await until(() => b.l1Size("inv") === 0);
      expect(await b.fetch("inv", "y", p, async () => 0)).toBe(21);
      // 断开 B 的监听连接:L1 停用;自动重连后重新启用且是空的
      expect(b.l1Size("inv")).toBe(1);
      await sql`select pg_terminate_backend(pid) from pg_stat_activity
        where query ilike 'listen umt_cache_invalidate%' and pid <> pg_backend_pid()`.execute(db2);
      await until(() => !b.l1Enabled || b.l1Size("inv") === 0);
      expect(b.l1Size("inv")).toBe(0);
      await until(() => b.l1Enabled && lb.connected(), 10_000);
      expect(b.l1Size("inv")).toBe(0);
    } finally {
      await la.stop();
      await lb.stop();
    }
    expect(a.l1Enabled).toBe(false);
  });

  it("过期在宽限期内:先返回旧值、后台只刷一次;源失败继续用旧值并记警告", async () => {
    const warns: string[] = [];
    const c = new MetadataCache(new PgCache(db), { l1: false, logger: { warn: (_o, m) => warns.push(m) } });
    let n = 0;
    let gate: () => void = () => {};
    const slow = () =>
      new Promise<number>((ok) => {
        n++;
        gate = () => ok(100 + n);
      });
    await c.fetch("swr", "k", { ttl: 0.2, grace: 60 }, async () => 1);
    await new Promise((r) => setTimeout(r, 300));
    const p = { ttl: 600 };
    const [x, y] = await Promise.all([c.fetch("swr", "k", p, slow), c.fetch("swr", "k", p, slow)]);
    expect([x, y]).toEqual([1, 1]);
    await until(() => n === 1);
    gate();
    await until(async () => (await new PgCache(db).get("swr", "k")) === 101);
    expect(n).toBe(1);
    expect(c.pendingStats().get("swr")).toMatchObject({ staleHits: 2, refreshes: 1 });
    // 再次过期,源失败:返回旧值,不删旧值,记一次警告
    await c.fetch("swr", "k2", { ttl: 0.2, grace: 60 }, async () => "old");
    await new Promise((r) => setTimeout(r, 300));
    const failing = async (): Promise<string> => {
      throw new Error("HTTP 503");
    };
    expect(await c.fetch("swr", "k2", p, failing)).toBe("old");
    await until(() => warns.length === 1);
    expect(warns[0]).toContain("继续用旧值");
    expect(await c.fetch("swr", "k2", p, async () => "new")).toBe("old");
    // 过了宽限期 = 硬过期,回源
    await c.fetch("swr", "k3", { ttl: 0.1, grace: 0.1 }, async () => "a");
    await new Promise((r) => setTimeout(r, 300));
    expect(await c.fetch("swr", "k3", p, async () => "b")).toBe("b");
    // PgCache.get 只认新鲜条目
    await c.fetch("swr", "k4", { ttl: 0.1, grace: 60 }, async () => "x");
    await new Promise((r) => setTimeout(r, 200));
    expect(await new PgCache(db).get("swr", "k4")).toBeNull();
    expect(await new PgCache(db).exists("swr", "k4")).toBe(true);
  });

  it("同请求合并:并发同 key 只回源一次;失败不缓存并计数", async () => {
    const c = new MetadataCache(new PgCache(db), { l1: true });
    let n = 0;
    const load = async () => {
      n++;
      await new Promise((r) => setTimeout(r, 50));
      return { ok: true };
    };
    const all = await Promise.all(Array.from({ length: 5 }, () => c.fetch("sf", "k", { ttl: 60 }, load)));
    expect(n).toBe(1);
    expect(all.every((v) => v.ok)).toBe(true);
    let m = 0;
    const fail = async () => {
      m++;
      throw new Error("ECONNRESET");
    };
    await expect(c.fetch("sf", "bad", { ttl: 60 }, fail)).rejects.toThrow("ECONNRESET");
    await expect(c.fetch("sf", "bad", { ttl: 60 }, fail)).rejects.toThrow("ECONNRESET");
    expect(m).toBe(2);
    expect(await new PgCache(db).exists("sf", "bad")).toBe(false);
    expect(c.pendingStats().get("sf")).toMatchObject({ failures: 2 });
    // skipIf / null 不缓存;空结果按 emptyTtl
    await c.fetch("sf", "none", { ttl: 60 }, async () => null);
    expect(await new PgCache(db).exists("sf", "none")).toBe(false);
    await c.fetch("sf", "empty", { ttl: 3600, emptyTtl: 60, isEmpty: (v: number[]) => !v.length }, async () => []);
    const t = await sql<{ s: number }>`select extract(epoch from fresh_until - now())::int as s
      from kv_cache where region = 'sf' and key = 'empty'`.execute(db);
    expect(t.rows[0]!.s).toBeLessThanOrEqual(60);
  });

  it("pinned:按条数上限淘汰时跳过;L2 命中时续期", async () => {
    const pgc = new PgCache(db);
    const c = new MetadataCache(pgc, { l1: false });
    const tick = () => new Promise((r) => setTimeout(r, 5));
    await c.fetch("pin", "watched", { ttl: 100, maxsize: 2, pin: async () => true }, async () => "w");
    for (const k of ["a", "b", "c", "d"]) {
      await tick();
      await c.fetch("pin", k, { ttl: 100, maxsize: 2 }, async () => k);
    }
    expect((await pgc.items("pin")).map((i) => i.key).sort()).toEqual(["d", "watched"]);
    // 续期:新鲜条目命中且 pin 为真 → 过期时间推到 now + ttl(保留宽限长度)
    await sql`update kv_cache set fresh_until = now() + interval '10 seconds', expires_at = now() + interval '60 seconds'
      where region = 'pin' and key = 'watched'`.execute(db);
    expect(await c.fetch("pin", "watched", { ttl: 3600, pin: async () => true }, async () => "x")).toBe("w");
    const r = await sql<{ f: number; g: number }>`select extract(epoch from fresh_until - now())::int as f,
      extract(epoch from expires_at - fresh_until)::int as g from kv_cache where region = 'pin' and key = 'watched'`.execute(
      db,
    );
    expect(r.rows[0]!.f).toBeGreaterThan(3500);
    expect(r.rows[0]!.g).toBe(50);
  });
});