UHD_Media_Tool/apps/server/src/modules/pi/refresh.ts

import { PiUpstreamClient, type UpstreamClientOptions } from "@umt/agent";
import { type PiUpstreamRow, sql } from "@umt/db";
import type { ServerContext } from "../types.ts";
import { dbCredentialStore, setStatus, specOf } from "./store.ts";

// refreshPiTokens:对即将过期的 OAuth 上游提前刷新。刷新本身走 pi 的 getAuth(minOAuthValidityMs):
// pi 在我们的 CredentialStore.modify(pi_upstream 行锁)里双重检查过期时间,再调 provider 的 refresh 并写回轮换后的
// token —— Agent 运行时(worker)与测试端点(api)的按需刷新走的是同一条路径,不会双写。
// 刷新失败(refresh token 失效)→ status = expired + 待处理事项“上游 X 需要重新登录”;旧凭据保留。

export interface RefreshOptions {
  /** 剩余有效期少于它就刷新(默认 30 分钟) */
  windowMs?: number;
  client?: UpstreamClientOptions;
}

export interface RefreshReport {
  checked: number;
  refreshed: number;
  failed: { upstreamId: string; message: string }[];
}

const DECISION_KEY = (id: string) => `pi_upstream_expired:${id}`;

/** 与其它模块的 upsertDecision 同一写法(跨模块不 import):同键未关闭时只累加 attempts */
async function upsertExpiredDecision(ctx: ServerContext, row: PiUpstreamRow, message: string) {
  const title = `Agent 上游「${row.name}」登录已失效,需要重新登录`;
  await ctx.db.transaction().execute(async (trx) => {
    const r = await sql<{ id: string; inserted: boolean }>`
      insert into decision (kind, severity, title, summary, subject_type, subject_id, dedupe_key, evidence, options)
      values ('other', 'warning', ${title}, ${message}, 'pi_upstream', ${row.id}, ${DECISION_KEY(row.id)},
              ${JSON.stringify({ provider: row.provider, upstreamId: row.id, error: message })}::jsonb,
              ${JSON.stringify([{ key: "relogin", label: "重新登录", description: "设置 → Agent → 上游" }])}::jsonb)
      on conflict (dedupe_key) where status in ('open', 'investigating', 'awaiting_user')
      do update set attempts = decision.attempts + 1, evidence = excluded.evidence, summary = excluded.summary,
                    updated_at = now()
      returning id, (xmax = 0) as inserted`.execute(trx);
    const d = r.rows[0]!;
    if (d.inserted)
      await ctx.outbox.emit(
        trx,
        "decision.created",
        { decisionId: d.id, kind: "other", severity: "warning", title },
        { idempotencyKey: `decision.created:${d.id}` },
      );
  });
}

/** 重新登录成功后关闭“需要重新登录”事项 */
export async function resolveExpiredDecision(ctx: ServerContext, upstreamId: string) {
  await ctx.db
    .updateTable("decision")
    .set({
      status: "resolved",
      resolved_at: new Date(),
      resolved_by: "system:pi",
      resolution: JSON.stringify({ key: "relogin" }),
    })
    .where("dedupe_key", "=", DECISION_KEY(upstreamId))
    .where("status", "in", ["open", "investigating", "awaiting_user"])
    .execute();
}

/** 刷新单个上游(api 测试端点与定时任务共用) */
export async function refreshUpstream(
  ctx: ServerContext,
  row: PiUpstreamRow,
  opts: RefreshOptions = {},
): Promise<{ ok: true } | { ok: false; message: string; kind: string }> {
  const client = new PiUpstreamClient(specOf(row), dbCredentialStore(ctx, row.id), opts.client);
  try {
    await client.resolveAuth(opts.windowMs ?? 30 * 60_000);
    return { ok: true };
  } catch (err) {
    const e = err as { kind?: string; message?: string };
    const message = e.message ?? String(err);
    if (e.kind === "oauth") {
      await setStatus(ctx, row.id, { status: "expired", lastError: message });
      await upsertExpiredDecision(ctx, row, message);
    } else if (e.kind === "auth") {
      await setStatus(ctx, row.id, { status: "auth_required", lastError: message });
    }
    return { ok: false, message, kind: e.kind ?? "other" };
  }
}

export async function refreshPiTokens(ctx: ServerContext, opts: RefreshOptions = {}): Promise<RefreshReport> {
  const windowMs = opts.windowMs ?? 30 * 60_000;
  const rows = await ctx.db
    .selectFrom("pi_upstream")
    .selectAll()
    .where("auth_kind", "=", "oauth")
    .where("enabled", "=", true)
    .where("secret_ref", "is not", null)
    .where("status", "!=", "expired")
    .where("oauth_expires_at", "is not", null)
    .where("oauth_expires_at", "<", new Date(Date.now() + windowMs))
    .execute();
  const report: RefreshReport = { checked: rows.length, refreshed: 0, failed: [] };
  for (const row of rows) {
    const r = await refreshUpstream(ctx, row, { ...opts, windowMs });
    if (r.ok) report.refreshed += 1;
    else {
      report.failed.push({ upstreamId: row.id, message: r.message });
      ctx.logger.warn({ upstreamId: row.id, provider: row.provider, message: r.message }, "Agent 上游 token 刷新失败");
    }
  }
  return report;
}