import { type Kysely, sql } from "kysely";
import pg from "pg";
import type { DB } from "./types.ts";

/**
 * PG 缓存后端:移植 MP app/runtime/cache.py 的内存后端(MemoryBackend + _MemoryTLRUCache)与 @cached 装饰器语义,
 * 只把存储从进程内存换成 PG UNLOGGED 表 kv_cache(迁移 0042),三进程共享。行为对照(docs/porting/moviepilot-tmdb-cache.md):
 * - region 分区;set 时每个 key 独立 TTL(秒),ttl ≤ 0 = 删除该 key;未给 ttl 用后端缺省(DEFAULT_CACHE_TTL 365 天);
 * - maxsize 按 region 计条数,超出时先删已过期、再删最久未用(TLRU);get 命中刷新"最近使用";
 * - get 未命中 / 已过期返回 null;clear(region) 清一个区;items(region) 列出未过期条目;
 * - cached():skipNone / skipEmpty / skipIf 不写缓存,emptyTtl + emptyIf 空结果短 TTL,fresh 跳过读缓存。
 * 与 MP 的差异只在存储介质:值用 JSON(jsonb)而不是 pickle;过期行由定时任务 sweep() 物理删除。
 */

/** MP DEFAULT_CACHE_TTL(秒) */
export const DEFAULT_CACHE_TTL = 365 * 24 * 60 * 60;
/** MP DEFAULT_CACHE_SIZE */
export const DEFAULT_CACHE_SIZE = 1024;

// biome-ignore lint/suspicious/noExplicitAny: 测试会传入隔离 schema 的实例
type AnyDb = Kysely<DB> | Kysely<any>;

export interface SetOptions {
  /** 秒;缺省 DEFAULT_CACHE_TTL;≤ 0 删除该 key */
  ttl?: number;
  /** region 条数上限;缺省 DEFAULT_CACHE_SIZE */
  maxsize?: number;
  /**
   * 宽限期(秒,0086):> 0 时 fresh_until = now + ttl、expires_at = now + ttl + grace,
   * 宽限期内的条目由 MetadataCache 先返回旧值再后台刷新;PgCache.get 只认新鲜条目。缺省 0。
   */
  grace?: number;
  /** 关注中作品的条目:按条数上限淘汰时跳过(到期照样过期)。缺省 false */
  pinned?: boolean;
}

/** 跨进程 L1 失效通知频道(payload 见 invalidationPayload) */
export const CACHE_INVALIDATE_CHANNEL = "umt_cache_invalidate";

/** 通知内容:`{"r":region,"k":key}`;k 缺省 = 整个 region;r 为 "*" = 全部。超长 key 退化为整 region */
function invalidationPayload(region: string, key?: string): string {
  const full = JSON.stringify(key === undefined ? { r: region } : { r: region, k: key });
  return full.length > 7000 ? JSON.stringify({ r: region }) : full;
}

/** getEntry 的结果:stale = 已过新鲜期、还在宽限期内 */
export interface CacheEntry<T> {
  value: T;
  stale: boolean;
  /** 新鲜期截止(毫秒);没有宽限期时同 expiresAt */
  freshUntil: number;
  /** 硬过期(毫秒) */
  expiresAt: number;
  pinned: boolean;
}

export class PgCache {
  constructor(readonly db: AnyDb) {}

  /** MemoryBackend.set */
  async set(region: string, key: string, value: unknown, opts: SetOptions = {}): Promise<void> {
    const ttl = opts.ttl ?? DEFAULT_CACHE_TTL;
    if (ttl <= 0) {
      await this.delete(region, key);
      return;
    }
    const grace = Math.max(0, opts.grace ?? 0);
    await sql`insert into kv_cache (region, key, value, expires_at, fresh_until, pinned, updated_at, accessed_at)
      values (${region}, ${key}, ${JSON.stringify(value ?? null)}::jsonb,
        now() + make_interval(secs => ${ttl + grace}),
        ${grace > 0 ? sql`now() + make_interval(secs => ${ttl})` : sql`null`},
        ${opts.pinned === true}, now(), now())
      on conflict (region, key) do update set value = excluded.value, expires_at = excluded.expires_at,
        fresh_until = excluded.fresh_until, pinned = excluded.pinned,
        updated_at = excluded.updated_at, accessed_at = excluded.accessed_at`.execute(this.db);
    await this.evict(region, opts.maxsize ?? DEFAULT_CACHE_SIZE);
  }

  /**
   * TLRUCache 写入时的淘汰:先删已过期,再按最近使用从旧到新删到 maxsize。
   * 0086:pinned(关注中作品)的条目不参与 LRU,但计入条数;刚写入的那条总留着。
   */
  private async evict(region: string, maxsize: number): Promise<void> {
    await sql`delete from kv_cache where region = ${region} and expires_at <= now()`.execute(this.db);
    await sql`delete from kv_cache where region = ${region} and not pinned and key in (
        select key from kv_cache where region = ${region} and not pinned
         order by accessed_at desc, updated_at desc
        offset greatest(1, ${Math.floor(maxsize)} - (select count(*) from kv_cache where region = ${region} and pinned)))`.execute(
      this.db,
    );
  }

  /** MemoryBackend.get:未命中 / 过期 / 已过新鲜期返回 null;命中刷新最近使用时间 */
  async get<T = unknown>(region: string, key: string): Promise<T | null> {
    const r = await sql<{ value: T }>`update kv_cache set accessed_at = now()
      where region = ${region} and key = ${key} and expires_at > now()
        and (fresh_until is null or fresh_until > now()) returning value`.execute(this.db);
    return r.rows.length ? (r.rows[0]!.value as T) : null;
  }

  /** 读一条(含宽限期内的旧值),命中刷新最近使用时间;MetadataCache 的 L2 读 */
  async getEntry<T = unknown>(region: string, key: string): Promise<CacheEntry<T> | null> {
    const r = await sql<{ value: T; expires_at: Date; fresh_until: Date | null; pinned: boolean; now: Date }>`
      update kv_cache set accessed_at = now()
       where region = ${region} and key = ${key} and expires_at > now()
       returning value, expires_at, fresh_until, pinned, now() as now`.execute(this.db);
    const row = r.rows[0];
    if (!row) return null;
    const expiresAt = new Date(row.expires_at).getTime();
    const freshUntil = row.fresh_until ? new Date(row.fresh_until).getTime() : expiresAt;
    return {
      value: row.value as T,
      stale: freshUntil <= new Date(row.now).getTime(),
      freshUntil,
      expiresAt,
      pinned: !!row.pinned,
    };
  }

  /**
   * 续期:新鲜的条目把过期时间推到 now + ttl(只延长不缩短,有宽限期的保留原宽限长度);返回是否有行被续期。
   * opts.pin = true 时顺带标为 pinned。旧值(已过新鲜期)不续期:该刷新就刷新。
   */
  async renew(region: string, key: string, ttl: number, opts: { pin?: boolean } = {}): Promise<boolean> {
    if (ttl <= 0) return false;
    const r = await sql`update kv_cache set
        expires_at = greatest(expires_at,
          now() + make_interval(secs => ${ttl}) + coalesce(expires_at - fresh_until, interval '0')),
        fresh_until = case when fresh_until is null then null
          else greatest(fresh_until, now() + make_interval(secs => ${ttl})) end,
        pinned = pinned or ${opts.pin === true}
      where region = ${region} and key = ${key} and expires_at > now()
        and (fresh_until is null or fresh_until > now())`.execute(this.db);
    return Number(r.numAffectedRows ?? 0) > 0;
  }

  /** 通知各进程丢掉 L1(删除 / 清空 / 手动刷新时);失败不抛 */
  async notifyInvalidate(region: string, key?: string): Promise<void> {
    await sql`select pg_notify(${CACHE_INVALIDATE_CHANNEL}, ${invalidationPayload(region, key)})`
      .execute(this.db)
      .catch(() => {});
  }

  async exists(region: string, key: string): Promise<boolean> {
    const r =
      await sql`select 1 from kv_cache where region = ${region} and key = ${key} and expires_at > now()`.execute(
        this.db,
      );
    return r.rows.length > 0;
  }

  /** 删除并通知各进程丢掉这条 L1 */
  async delete(region: string, key: string): Promise<void> {
    await sql`delete from kv_cache where region = ${region} and key = ${key}`.execute(this.db);
    await this.notifyInvalidate(region, key);
  }

  /** 清一个区,返回删除行数;通知各进程丢掉这个区的 L1 */
  async clear(region: string): Promise<number> {
    const r = await sql`delete from kv_cache where region = ${region}`.execute(this.db);
    await this.notifyInvalidate(region);
    return Number(r.numAffectedRows ?? 0);
  }

  /** MemoryBackend.items:未过期条目(附过期时间与大小,管理界面用),按 key 排序 */
  async items<T = unknown>(
    region: string,
    limit = 5000,
  ): Promise<{ key: string; value: T; expiresAt: Date; freshUntil: Date; updatedAt: Date; bytes: number }[]> {
    const r = await sql<{
      key: string;
      value: T;
      expires_at: Date;
      fresh_until: Date;
      updated_at: Date;
      bytes: number;
    }>`
      select key, value, expires_at, coalesce(fresh_until, expires_at) as fresh_until, updated_at,
             pg_column_size(value)::int as bytes
        from kv_cache where region = ${region} and expires_at > now() order by key limit ${limit}`.execute(this.db);
    return r.rows.map((x) => ({
      key: x.key,
      value: x.value,
      expiresAt: x.expires_at,
      /** 新鲜期截止(有宽限期的条目早于 expiresAt) */
      freshUntil: x.fresh_until,
      updatedAt: x.updated_at,
      bytes: Number(x.bytes),
    }));
  }

  /** 未过期条目数(len(region_cache)) */
  async count(region: string): Promise<number> {
    const r = await sql<{ n: number }>`select count(*)::int as n from kv_cache
      where region = ${region} and expires_at > now()`.execute(this.db);
    return Number(r.rows[0]?.n ?? 0);
  }

  /** 删除全部过期行(存储换成 PG 后需要的物理清理;cache.sweep 定时任务) */
  async sweep(): Promise<number> {
    const r = await sql`delete from kv_cache where expires_at <= now()`.execute(this.db);
    return Number(r.numAffectedRows ?? 0);
  }
}

/** MP @cached 的参数 */
export interface CachedOptions<T> {
  region: string;
  maxsize?: number;
  /** 秒 */
  ttl?: number;
  /** 缺省 true:结果为 null / undefined 不缓存 */
  skipNone?: boolean;
  /** 缺省 false:结果为假值(空数组 / 空对象 / 空串…)不缓存 */
  skipEmpty?: boolean;
  /** 返回真值时不缓存(如业务失败的响应) */
  skipIf?: (v: T) => boolean;
  /** 空结果的独立 TTL(秒) */
  emptyTtl?: number;
  /** 判断空结果;缺省按假值 */
  emptyIf?: (v: T) => boolean;
  /** MP fresh():true 时不读缓存(仍写) */
  fresh?: boolean;
}

const isFalsy = (v: unknown) =>
  !v || (Array.isArray(v) && v.length === 0) || (typeof v === "object" && Object.keys(v as object).length === 0);

/** MP @cached 的调用体:命中直接返回;否则执行 fn,按 should_cache / get_cache_ttl 写入 */
export async function cachedCall<T>(
  cache: PgCache,
  key: string,
  o: CachedOptions<T>,
  fn: () => Promise<T>,
): Promise<T> {
  const should = (v: T | null | undefined) => {
    if ((o.skipNone ?? true) && (v === null || v === undefined)) return false;
    if (o.skipEmpty && isFalsy(v)) return false;
    if (o.skipIf && v !== null && v !== undefined && o.skipIf(v)) return false;
    return true;
  };
  if (!o.fresh) {
    const hit = await cache.get<T>(o.region, key);
    if (should(hit)) return hit as T;
  }
  const result = await fn();
  if (!should(result)) return result;
  let ttl = o.ttl;
  if (o.emptyTtl !== undefined) {
    if (result === null || result === undefined) ttl = o.emptyTtl;
    else if (o.emptyIf) ttl = o.emptyIf(result) ? o.emptyTtl : o.ttl;
    else ttl = isFalsy(result) ? o.emptyTtl : o.ttl;
  }
  await cache.set(o.region, key, result, {
    ...(ttl !== undefined ? { ttl } : {}),
    ...(o.maxsize ? { maxsize: o.maxsize } : {}),
  });
  return result;
}

const caches = new WeakMap<object, PgCache>();

/** 同一个 Kysely 实例共用一个 PgCache */
export function pgCacheFor(db: AnyDb): PgCache {
  let c = caches.get(db);
  if (!c) {
    c = new PgCache(db);
    caches.set(db, c);
  }
  return c;
}

// ── 统一元数据缓存(TMDB / 豆瓣 / Bangumi 共用;docs/runbook/cache.md) ──
// 两级:L1 进程内存(按 region 限条数的 LRU,TTL 短)+ L2 PG kv_cache(三进程共享)。读 L1 → L2 → 源,写两级都写。
// L1 只在跨进程失效通知(LISTEN umt_cache_invalidate)在线时启用:离线或断线重连时整体清空并停用,
// 避免别的进程清空 / 删除后本进程还拿着旧值。

/** 宽限期:TTL 的 50%,上限 7 天 */
export const STALE_GRACE_RATIO = 0.5;
export const STALE_GRACE_MAX_SEC = 7 * 86_400;
export const defaultGrace = (ttlSec: number) => Math.min(Math.max(0, ttlSec) * STALE_GRACE_RATIO, STALE_GRACE_MAX_SEC);

export interface FetchPolicy<T> {
  /** 秒 */
  ttl: number;
  /** 空结果 TTL(秒);缺省同 ttl */
  emptyTtl?: number;
  /** 空结果判定;缺省按假值(null / 空数组 / 空对象) */
  isEmpty?: (v: T) => boolean;
  maxsize?: number;
  /** 宽限期(秒);缺省 defaultGrace(本条 TTL);0 = 过期就回源 */
  grace?: number;
  /** 缺省 true:null / undefined 不缓存 */
  skipNone?: boolean;
  /** 返回真值时不缓存(业务失败的响应) */
  skipIf?: (v: T) => boolean;
  /** true:跳过读(手动刷新),照常写,并通知各进程丢掉这条 L1 */
  fresh?: boolean;
  /**
   * 关注中作品的条目:写入时标 pinned(LRU 不淘汰),L2 命中时按 ttl 续期。
   * 只在写入 / L2 命中时调用,失败按 false。
   */
  pin?: () => Promise<boolean>;
}

export interface CacheCounters {
  l1Hits: number;
  l2Hits: number;
  /** 先返回旧值(宽限期内)的次数 */
  staleHits: number;
  misses: number;
  /** 回源失败(网络、被拦、5xx 等抛错) */
  failures: number;
  /** 后台刷新次数 */
  refreshes: number;
}

const zero = (): CacheCounters => ({ l1Hits: 0, l2Hits: 0, staleHits: 0, misses: 0, failures: 0, refreshes: 0 });

export interface MetadataCacheOptions {
  /** L1 每个 region 的条数上限,缺省 1000 */
  l1MaxPerRegion?: number;
  /** L1 TTL(毫秒),缺省 5 分钟;不超过条目在 L2 的新鲜期 */
  l1TtlMs?: number;
  /** 是否立即启用 L1(测试用;服务端由失效监听上线时打开) */
  l1?: boolean;
  now?: () => number;
  logger?: { warn(obj: object, msg: string): void };
}

interface L1Entry {
  value: unknown;
  freshUntil: number;
  /** L1 内的过期(min(写入 + l1Ttl, 硬过期)) */
  expiresAt: number;
}

const clone = <T>(v: T): T => (v === null || typeof v !== "object" ? v : structuredClone(v));

export class MetadataCache {
  private readonly l1 = new Map<string, Map<string, L1Entry>>();
  private readonly inflight = new Map<string, Promise<unknown>>();
  private readonly counters = new Map<string, CacheCounters>();
  private l1On: boolean;
  private readonly now: () => number;

  constructor(
    readonly pg: PgCache,
    private readonly opts: MetadataCacheOptions = {},
  ) {
    this.l1On = opts.l1 ?? false;
    this.now = opts.now ?? Date.now;
  }

  // ── L1 ──

  get l1Enabled(): boolean {
    return this.l1On;
  }

  /** 打开 / 关闭 L1;状态变化时都整体清空 */
  setL1Enabled(on: boolean): void {
    if (this.l1On !== on) this.l1.clear();
    this.l1On = on;
  }

  /** 丢掉本进程 L1:给 key 丢一条,不给丢整个 region,region 为 "*" 全部丢 */
  dropLocal(region: string, key?: string): void {
    if (region === "*") this.l1.clear();
    else if (key === undefined) this.l1.delete(region);
    else this.l1.get(region)?.delete(key);
  }

  /** L1 当前条数(测试 / 诊断) */
  l1Size(region: string): number {
    return this.l1.get(region)?.size ?? 0;
  }

  private l1Get(region: string, key: string): L1Entry | null {
    if (!this.l1On) return null;
    const m = this.l1.get(region);
    const e = m?.get(key);
    if (!m || !e) return null;
    if (e.expiresAt <= this.now()) {
      m.delete(key);
      return null;
    }
    m.delete(key);
    m.set(key, e);
    return e;
  }

  private l1Set(region: string, key: string, value: unknown, freshUntil: number, hardExpiresAt: number): void {
    if (!this.l1On) return;
    let m = this.l1.get(region);
    if (!m) {
      m = new Map();
      this.l1.set(region, m);
    }
    m.delete(key);
    m.set(key, {
      value: clone(value),
      freshUntil,
      expiresAt: Math.min(this.now() + (this.opts.l1TtlMs ?? 5 * 60_000), hardExpiresAt),
    });
    const max = this.opts.l1MaxPerRegion ?? 1000;
    while (m.size > max) {
      const oldest = m.keys().next().value;
      if (oldest === undefined) break;
      m.delete(oldest);
    }
  }

  // ── 统计 ──

  private count(region: string, k: keyof CacheCounters, n = 1): void {
    let c = this.counters.get(region);
    if (!c) {
      c = zero();
      this.counters.set(region, c);
    }
    c[k] += n;
  }

  /** 本进程还没写进 cache_stat 的计数 */
  pendingStats(): Map<string, CacheCounters> {
    return new Map([...this.counters].map(([r, c]) => [r, { ...c }]));
  }

  /**
   * 把本进程的计数增量加进 cache_stat(按天 × region),成功后清零;失败保留到下次。
   * 顺带删掉 30 天前的统计行。返回写入的 region 数。
   */
  async flushStats(): Promise<number> {
    const batch = [...this.counters];
    if (!batch.length) return 0;
    this.counters.clear();
    let done = 0;
    try {
      for (const [region, c] of batch) {
        await sql`insert into cache_stat (day, region, l1_hits, l2_hits, stale_hits, misses, failures, refreshes, updated_at)
          values (current_date, ${region}, ${c.l1Hits}, ${c.l2Hits}, ${c.staleHits}, ${c.misses}, ${c.failures}, ${c.refreshes}, now())
          on conflict (day, region) do update set
            l1_hits = cache_stat.l1_hits + excluded.l1_hits, l2_hits = cache_stat.l2_hits + excluded.l2_hits,
            stale_hits = cache_stat.stale_hits + excluded.stale_hits, misses = cache_stat.misses + excluded.misses,
            failures = cache_stat.failures + excluded.failures, refreshes = cache_stat.refreshes + excluded.refreshes,
            updated_at = now()`.execute(this.pg.db);
        done++;
      }
      await sql`delete from cache_stat where day < current_date - 30`.execute(this.pg.db);
      return done;
    } catch (err) {
      // 没写进去的加回去
      for (const [region, c] of batch.slice(done))
        for (const k of Object.keys(c) as (keyof CacheCounters)[]) this.count(region, k, c[k]);
      throw err;
    }
  }

  private warn(obj: object, msg: string) {
    this.opts.logger?.warn(obj, msg);
  }

  // ── 读写 ──

  /** 只读(不回源):L1 → L2 新鲜条目;记命中 / 未命中 */
  async get<T = unknown>(region: string, key: string): Promise<T | null> {
    const l1 = this.l1Get(region, key);
    if (l1 && l1.freshUntil > this.now()) {
      this.count(region, "l1Hits");
      return clone(l1.value) as T;
    }
    const e = await this.pg.getEntry<T>(region, key).catch((err) => {
      this.warn({ region, err: { message: (err as Error).message } }, "缓存读取失败，按未命中处理");
      return null;
    });
    if (!e || e.stale || e.value === null || e.value === undefined) {
      this.count(region, "misses");
      return null;
    }
    this.count(region, "l2Hits");
    this.l1Set(region, key, e.value, e.freshUntil, e.expiresAt);
    return e.value;
  }

  /** 直接写两级(ttl ≤ 0 删除) */
  async set(region: string, key: string, value: unknown, opts: SetOptions = {}): Promise<void> {
    await this.pg.set(region, key, value, opts);
    const ttl = opts.ttl ?? DEFAULT_CACHE_TTL;
    if (ttl <= 0) {
      this.dropLocal(region, key);
      return;
    }
    const fresh = this.now() + ttl * 1000;
    this.l1Set(region, key, value, fresh, fresh + Math.max(0, opts.grace ?? 0) * 1000);
  }

  async delete(region: string, key: string): Promise<void> {
    this.dropLocal(region, key);
    await this.pg.delete(region, key);
  }

  async clear(region: string): Promise<number> {
    this.dropLocal(region);
    return await this.pg.clear(region);
  }

  /**
   * 带回源的读(MP @cached 的调用体 + 两级 + 过期先用旧值 + 同请求合并 + 统计):
   * - L1 / L2 新鲜命中直接返回;L2 命中且 pin() 为真时续期;
   * - 宽限期内的旧值先返回,后台刷新(同 key 同时只刷一次),刷新失败继续用旧值并记一次警告;
   * - 未命中:同进程同 key 只发一次 load;load 抛错不写缓存(网络、被拦、5xx);结果按 skipNone / skipIf 决定是否写。
   * PG 读写失败按未命中 / 不写处理,不让业务请求失败。
   */
  async fetch<T>(region: string, key: string, p: FetchPolicy<T>, load: () => Promise<T>): Promise<T> {
    if (!p.fresh) {
      const l1 = this.l1Get(region, key);
      if (l1) {
        if (l1.freshUntil > this.now()) {
          this.count(region, "l1Hits");
          return clone(l1.value) as T;
        }
        this.count(region, "staleHits");
        this.revalidate(region, key, p, load);
        return clone(l1.value) as T;
      }
      const e = await this.pg.getEntry<T>(region, key).catch((err) => {
        this.warn({ region, err: { message: (err as Error).message } }, "缓存读取失败，按未命中处理");
        return null;
      });
      if (e && this.shouldCache(p, e.value)) {
        this.l1Set(region, key, e.value, e.freshUntil, e.expiresAt);
        if (!e.stale) {
          this.count(region, "l2Hits");
          if (p.pin) await this.renewPinned(region, key, p);
          return e.value;
        }
        this.count(region, "staleHits");
        this.revalidate(region, key, p, load);
        return e.value;
      }
    }
    this.count(region, "misses");
    return (await this.loadOnce(region, key, p, load)) as T;
  }

  private shouldCache<T>(p: FetchPolicy<T>, v: T | null | undefined): v is T {
    if ((p.skipNone ?? true) && (v === null || v === undefined)) return false;
    if (p.skipIf && v !== null && v !== undefined && p.skipIf(v)) return false;
    return true;
  }

  private async renewPinned<T>(region: string, key: string, p: FetchPolicy<T>): Promise<void> {
    try {
      if (await p.pin!()) await this.pg.renew(region, key, p.ttl, { pin: true });
    } catch (err) {
      this.warn({ region, err: { message: (err as Error).message } }, "缓存续期失败");
    }
  }

  private loadOnce<T>(region: string, key: string, p: FetchPolicy<T>, load: () => Promise<T>): Promise<T> {
    const id = `${region}\u0000${key}`;
    const pending = this.inflight.get(id);
    if (pending) return pending as Promise<T>;
    const run = (async () => {
      let v: T;
      try {
        v = await load();
      } catch (err) {
        this.count(region, "failures");
        throw err;
      }
      if (this.shouldCache(p, v)) await this.write(region, key, p, v);
      return v;
    })().finally(() => this.inflight.delete(id));
    this.inflight.set(id, run);
    return run;
  }

  /** 宽限期内的后台刷新:同 key 已在刷新 / 加载就不再发 */
  private revalidate<T>(region: string, key: string, p: FetchPolicy<T>, load: () => Promise<T>): void {
    if (this.inflight.has(`${region}\u0000${key}`)) return;
    this.count(region, "refreshes");
    this.loadOnce(region, key, { ...p, fresh: false }, load).catch((err) =>
      this.warn({ region, key, err: { message: (err as Error).message } }, "缓存后台刷新失败，继续用旧值"),
    );
  }

  private async write<T>(region: string, key: string, p: FetchPolicy<T>, v: T): Promise<void> {
    const empty = p.isEmpty ? p.isEmpty(v) : isFalsy(v);
    const ttl = empty && p.emptyTtl !== undefined ? p.emptyTtl : p.ttl;
    const grace = p.grace ?? defaultGrace(ttl);
    let pinned = false;
    if (p.pin) pinned = await p.pin().catch(() => false);
    try {
      await this.pg.set(region, key, v, { ttl, grace, pinned, ...(p.maxsize ? { maxsize: p.maxsize } : {}) });
    } catch (err) {
      this.warn({ region, err: { message: (err as Error).message } }, "缓存写入失败");
    }
    const fresh = this.now() + ttl * 1000;
    this.l1Set(region, key, v, fresh, fresh + grace * 1000);
    if (p.fresh) await this.pg.notifyInvalidate(region, key);
  }
}

const metadataCaches = new WeakMap<object, MetadataCache>();

/** 同一个 Kysely 实例共用一个 MetadataCache(L2 用 pgCacheFor(db)) */
export function metadataCacheFor(db: AnyDb, opts?: MetadataCacheOptions): MetadataCache {
  let c = metadataCaches.get(db);
  if (!c) {
    c = new MetadataCache(pgCacheFor(db), opts);
    metadataCaches.set(db, c);
  }
  return c;
}

export interface InvalidationListenerOptions {
  connectionString: string;
  logger?: { warn(obj: object, msg: string): void; info(obj: object, msg: string): void };
  /** 断线重连退避上限,缺省 10s */
  maxReconnectMs?: number;
}

/**
 * 跨进程 L1 失效:一条专用连接 LISTEN umt_cache_invalidate,收到通知丢掉对应 L1。
 * 连上时清空并启用 L1;断开时停用(清空)L1 并退避重连,重连成功再整体清一次后启用。
 */
export async function listenCacheInvalidation(
  cache: MetadataCache,
  opts: InvalidationListenerOptions,
): Promise<{ stop(): Promise<void>; connected(): boolean }> {
  let client: pg.Client | null = null;
  let stopped = false;
  let timer: ReturnType<typeof setTimeout> | null = null;

  const connect = async () => {
    const c = new pg.Client({ connectionString: opts.connectionString });
    c.on("notification", (msg) => {
      if (msg.channel !== CACHE_INVALIDATE_CHANNEL) return;
      try {
        const p = JSON.parse(msg.payload ?? "{}") as { r?: string; k?: string };
        if (typeof p.r === "string") cache.dropLocal(p.r, typeof p.k === "string" ? p.k : undefined);
      } catch {
        cache.dropLocal("*");
      }
    });
    const lost = (err?: Error) => {
      if (client !== c) return;
      client = null;
      cache.setL1Enabled(false);
      c.removeAllListeners("notification");
      c.end().catch(() => {});
      if (!stopped) {
        opts.logger?.warn({ err: err ? { message: err.message } : undefined }, "缓存失效监听断开，L1 停用并准备重连");
        reconnect(500);
      }
    };
    c.on("error", lost);
    c.on("end", () => lost());
    await c.connect();
    await c.query(`listen ${CACHE_INVALIDATE_CHANNEL}`);
    client = c;
    cache.setL1Enabled(false);
    cache.setL1Enabled(true);
  };
  const reconnect = (delay: number) => {
    if (stopped || timer) return;
    timer = setTimeout(async () => {
      timer = null;
      try {
        await connect();
        opts.logger?.info({}, "缓存失效监听已重连");
      } catch (err) {
        opts.logger?.warn({ err: { message: (err as Error).message } }, "缓存失效监听重连失败");
        reconnect(Math.min(delay * 2, opts.maxReconnectMs ?? 10_000));
      }
    }, delay);
  };

  await connect();
  return {
    connected: () => client !== null,
    async stop() {
      stopped = true;
      if (timer) clearTimeout(timer);
      const c = client;
      client = null;
      cache.setL1Enabled(false);
      if (c) {
        c.removeAllListeners("notification");
        await c.end().catch(() => {});
      }
    },
  };
}