[codex] add sub2api managed refresh resilience (#441)

* add sub2api managed refresh resilience * serialize sub2api managed refresh per account * bound sub2api refresh scheduler work * tighten sub2api refresh scheduler rules
2026-04-06 11:43:30 +08:00
parent 3edeb6e245
commit a789ddf7f8
8 changed files with 825 additions and 116 deletions
@@ -44,11 +44,11 @@ import {
  startChannelRecoveryProbeScheduler,
  stopChannelRecoveryProbeScheduler,
 } from './services/channelRecoveryProbeService.js';
-import { startUpdateCenterPolling, stopUpdateCenterPolling } from './services/updateCenterPollingService.js';
 import {
-  startOauthTokenRefreshScheduler,
-  stopOauthTokenRefreshScheduler,
-} from './services/oauth/oauthRefreshScheduler.js';
+  startSub2ApiManagedRefreshScheduler,
+  stopSub2ApiManagedRefreshScheduler,
+} from './services/sub2apiRefreshScheduler.js';
+import { startUpdateCenterPolling, stopUpdateCenterPolling } from './services/updateCenterPollingService.js';
 import { reloadBackupWebdavScheduler } from './services/backupService.js';
 import { ensureRuntimeDatabaseReady } from './runtimeDatabaseBootstrap.js';
 import { isPublicApiRoute, registerDesktopRoutes } from './desktop.js';
@@ -258,8 +258,8 @@ await reloadBackupWebdavScheduler();
 startSiteAnnouncementPolling();
 startModelAvailabilityProbeScheduler();
 startChannelRecoveryProbeScheduler();
+startSub2ApiManagedRefreshScheduler();
 startUpdateCenterPolling();
-startOauthTokenRefreshScheduler();
 try {
  await startOAuthLoopbackCallbackServers();
 } catch (error) {
@@ -274,7 +274,7 @@ app.addHook('onClose', async () => {
  stopProxyLogRetentionService();
  stopModelAvailabilityProbeScheduler();
  stopChannelRecoveryProbeScheduler();
-  await stopOauthTokenRefreshScheduler();
+  await stopSub2ApiManagedRefreshScheduler();
  await stopOAuthLoopbackCallbackServers();
 });

@@ -333,6 +333,53 @@ describe('balanceService auto relogin', () => {
    expect(reportTokenExpiredMock).not.toHaveBeenCalled();
  });

+  it('surfaces upstream sub2api refresh failure detail when managed refresh is rejected', async () => {
+    selectAllMock.mockReturnValue([
+      {
+        accounts: {
+          id: 8,
+          username: 'sub2-user',
+          accessToken: 'expired-access-token',
+          status: 'active',
+          extraConfig: JSON.stringify({
+            sub2apiAuth: {
+              refreshToken: 'refresh-token-invalid',
+            },
+          }),
+        },
+        sites: {
+          id: 9,
+          name: 'sub2',
+          url: 'https://sub2.example.com',
+          platform: 'sub2api',
+        },
+      },
+    ]);
+
+    adapterMock.getBalance.mockRejectedValueOnce(new Error('HTTP 401: unauthorized'));
+    undiciFetchMock.mockResolvedValueOnce({
+      ok: false,
+      status: 401,
+      text: async () => JSON.stringify({
+        code: 401,
+        message: 'invalid refresh token',
+        reason: 'REFRESH_TOKEN_INVALID',
+      }),
+      json: async () => ({
+        code: 401,
+        message: 'invalid refresh token',
+        reason: 'REFRESH_TOKEN_INVALID',
+      }),
+    });
+
+    const { refreshBalance } = await import('./balanceService.js');
+    await expect(refreshBalance(8)).rejects.toThrow(
+      'sub2api token refresh failed: HTTP 401: invalid refresh token (REFRESH_TOKEN_INVALID)',
+    );
+
+    expect(adapterMock.login).not.toHaveBeenCalled();
+  });
+
  it('skips balance refresh for api-key-only accounts without expiring them', async () => {
    selectAllMock.mockReturnValue([
      {
@@ -17,6 +17,11 @@ import { extractRuntimeHealth, setAccountRuntimeHealth } from './accountHealthSe
 import { updateTodayIncomeSnapshot } from './todayIncomeRewardService.js';
 import type { BalanceInfo } from './platforms/base.js';
 import { withAccountProxyOverride, withSiteProxyRequestInit, withSiteRecordProxyRequestInit } from './siteProxy.js';
+import {
+  isManagedSub2ApiTokenDue,
+  isSub2ApiPlatform,
+} from './sub2apiManagedAuth.js';
+import { refreshSub2ApiManagedSessionSingleflight } from './sub2apiRefreshSingleflight.js';

 function isSiteDisabled(status?: string | null): boolean {
  return (status || 'active') === 'disabled';
@@ -129,111 +134,6 @@ function resolveQuotaConversionFactor(platform?: string | null): number {
  return (platform || '').toLowerCase() === 'veloera' ? 1_000_000 : 500_000;
 }

-const SUB2API_TOKEN_REFRESH_BUFFER_MS = 120 * 1000;
-
-function isSub2ApiPlatform(platform?: string | null): boolean {
-  return (platform || '').trim().toLowerCase() === 'sub2api';
-}
-
-function isNearTokenExpiry(tokenExpiresAt?: number): boolean {
-  if (!(typeof tokenExpiresAt === 'number' && Number.isFinite(tokenExpiresAt) && tokenExpiresAt > 0)) {
-    return false;
-  }
-  return tokenExpiresAt - Date.now() <= SUB2API_TOKEN_REFRESH_BUFFER_MS;
-}
-
-type Sub2ApiRefreshedCredentials = {
-  accessToken: string;
-  refreshToken: string;
-  tokenExpiresAt: number;
-};
-
-function parseSub2ApiRefreshPayload(payload: unknown): Sub2ApiRefreshedCredentials | null {
-  if (!payload || typeof payload !== 'object' || Array.isArray(payload)) return null;
-  const body = payload as Record<string, unknown>;
-  const code = typeof body.code === 'number'
-    ? body.code
-    : (typeof body.code === 'string' ? Number.parseInt(body.code.trim(), 10) : NaN);
-  if (!Number.isFinite(code) || code !== 0) return null;
-  if (!body.data || typeof body.data !== 'object' || Array.isArray(body.data)) return null;
-  const data = body.data as Record<string, unknown>;
-
-  const accessToken = typeof data.access_token === 'string' ? data.access_token.trim() : '';
-  const refreshToken = typeof data.refresh_token === 'string' ? data.refresh_token.trim() : '';
-  const expiresInSeconds = typeof data.expires_in === 'number' && Number.isFinite(data.expires_in)
-    ? data.expires_in
-    : (typeof data.expires_in === 'string' ? Number.parseInt(data.expires_in.trim(), 10) : NaN);
-
-  if (!accessToken || !refreshToken || !Number.isFinite(expiresInSeconds) || expiresInSeconds <= 0) {
-    return null;
-  }
-
-  return {
-    accessToken,
-    refreshToken,
-    tokenExpiresAt: Date.now() + Math.trunc(expiresInSeconds) * 1000,
-  };
-}
-
-async function refreshSub2ApiManagedSession(params: {
-  account: typeof schema.accounts.$inferSelect;
-  site: typeof schema.sites.$inferSelect;
-  currentAccessToken: string;
-  currentExtraConfig: string | null;
-}): Promise<{ accessToken: string; extraConfig: string }> {
-  const managedAuth = getSub2ApiAuthFromExtraConfig(params.currentExtraConfig);
-  const refreshToken = managedAuth?.refreshToken || '';
-  if (!refreshToken) throw new Error('sub2api managed refresh token missing');
-
-  const endpoint = `${params.site.url.replace(/\/+$/, '')}/api/v1/auth/refresh`;
-  const headers: Record<string, string> = {
-    'Content-Type': 'application/json',
-  };
-  const authHeaderToken = (params.currentAccessToken || '').trim();
-  if (authHeaderToken) {
-    headers.Authorization = `Bearer ${authHeaderToken}`;
-  }
-
-  const { fetch } = await import('undici');
-  let payload: unknown = null;
-  try {
-    const response = await fetch(endpoint, withSiteRecordProxyRequestInit(params.site, {
-      method: 'POST',
-      headers,
-      body: JSON.stringify({ refresh_token: refreshToken }),
-    }, resolveProxyUrlFromExtraConfig(params.account.extraConfig)));
-    payload = await response.json().catch(() => null);
-  } catch (err: any) {
-    throw new Error(err?.message || 'sub2api token refresh request failed');
-  }
-
-  const refreshed = parseSub2ApiRefreshPayload(payload);
-  if (!refreshed) {
-    throw new Error('sub2api token refresh failed');
-  }
-
-  const nextExtraConfig = mergeAccountExtraConfig(params.currentExtraConfig, {
-    sub2apiAuth: {
-      refreshToken: refreshed.refreshToken,
-      tokenExpiresAt: refreshed.tokenExpiresAt,
-    },
-  });
-  await db.update(schema.accounts)
-    .set({
-      accessToken: refreshed.accessToken,
-      extraConfig: nextExtraConfig,
-      status: params.account.status === 'expired' ? 'active' : params.account.status,
-      updatedAt: new Date().toISOString(),
-    })
-    .where(eq(schema.accounts.id, params.account.id))
-    .run();
-
-  return {
-    accessToken: refreshed.accessToken,
-    extraConfig: nextExtraConfig,
-  };
-}
-
 async function fetchTodayIncomeFromLogs(params: {
  baseUrl: string;
  accessToken: string;
@@ -386,14 +286,14 @@ export async function refreshBalance(accountId: number) {

  if (isSub2ApiPlatform(site.platform)) {
    const managedAuth = getSub2ApiAuthFromExtraConfig(activeExtraConfig);
-    if (managedAuth?.refreshToken && isNearTokenExpiry(managedAuth.tokenExpiresAt)) {
+    if (managedAuth?.refreshToken && isManagedSub2ApiTokenDue(managedAuth.tokenExpiresAt)) {
      try {
-        const refreshed = await withAccountProxyOverride(accountProxyUrl, () => refreshSub2ApiManagedSession({
+        const refreshed = await refreshSub2ApiManagedSessionSingleflight({
          account,
          site,
          currentAccessToken: activeAccessToken,
          currentExtraConfig: activeExtraConfig,
-        }));
+        });
        activeAccessToken = refreshed.accessToken;
        activeExtraConfig = refreshed.extraConfig;
      } catch {}
@@ -430,12 +330,12 @@ export async function refreshBalance(accountId: number) {

    if (canTryManagedSub2ApiRefresh) {
      try {
-        const refreshed = await withAccountProxyOverride(accountProxyUrl, () => refreshSub2ApiManagedSession({
+        const refreshed = await refreshSub2ApiManagedSessionSingleflight({
          account,
          site,
          currentAccessToken: activeAccessToken,
          currentExtraConfig: activeExtraConfig,
-        }));
+        });
        activeAccessToken = refreshed.accessToken;
        activeExtraConfig = refreshed.extraConfig;
        balanceInfo = await readBalance(activeAccessToken);
@@ -0,0 +1,189 @@
+import { eq } from 'drizzle-orm';
+import { db, schema } from '../db/index.js';
+import {
+  getSub2ApiAuthFromExtraConfig,
+  mergeAccountExtraConfig,
+  resolveProxyUrlFromExtraConfig,
+} from './accountExtraConfig.js';
+import { withSiteRecordProxyRequestInit } from './siteProxy.js';
+
+export const SUB2API_MANAGED_REFRESH_LEAD_MS = 120 * 1000;
+
+type Sub2ApiRefreshedCredentials = {
+  accessToken: string;
+  refreshToken: string;
+  tokenExpiresAt: number;
+};
+
+function normalizeNonEmptyString(value: unknown): string {
+  if (typeof value !== 'string') return '';
+  return value.trim();
+}
+
+function normalizeErrorSnippet(value: string): string {
+  const normalized = value.replace(/\s+/g, ' ').trim();
+  if (!normalized) return '';
+  return normalized.length > 200
+    ? `${normalized.slice(0, 197)}...`
+    : normalized;
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return !!value && typeof value === 'object' && !Array.isArray(value);
+}
+
+function parseSub2ApiRefreshPayload(payload: unknown): Sub2ApiRefreshedCredentials | null {
+  if (!isRecord(payload)) return null;
+  const code = typeof payload.code === 'number'
+    ? payload.code
+    : (typeof payload.code === 'string' ? Number.parseInt(payload.code.trim(), 10) : Number.NaN);
+  if (!Number.isFinite(code) || code !== 0) return null;
+  if (!isRecord(payload.data)) return null;
+  const data = payload.data;
+
+  const accessToken = normalizeNonEmptyString(data.access_token);
+  const refreshToken = normalizeNonEmptyString(data.refresh_token);
+  const expiresInSeconds = typeof data.expires_in === 'number' && Number.isFinite(data.expires_in)
+    ? data.expires_in
+    : (typeof data.expires_in === 'string' ? Number.parseInt(data.expires_in.trim(), 10) : Number.NaN);
+
+  if (!accessToken || !refreshToken || !Number.isFinite(expiresInSeconds) || expiresInSeconds <= 0) {
+    return null;
+  }
+
+  return {
+    accessToken,
+    refreshToken,
+    tokenExpiresAt: Date.now() + Math.trunc(expiresInSeconds) * 1000,
+  };
+}
+
+function parseJsonPayload(rawText: string): unknown {
+  if (!rawText.trim()) return null;
+  try {
+    return JSON.parse(rawText) as unknown;
+  } catch {
+    return null;
+  }
+}
+
+function buildSub2ApiRefreshFailureMessage(input: {
+  status?: number;
+  rawText?: string;
+  payload?: unknown;
+}): string {
+  const prefix = typeof input.status === 'number' && Number.isFinite(input.status) && input.status > 0
+    ? `sub2api token refresh failed: HTTP ${Math.trunc(input.status)}`
+    : 'sub2api token refresh failed';
+
+  if (isRecord(input.payload)) {
+    const message = normalizeNonEmptyString(
+      input.payload.message
+      ?? input.payload.error
+      ?? input.payload.error_description,
+    );
+    const reason = normalizeNonEmptyString(input.payload.reason);
+    if (message && reason && reason.toLowerCase() !== message.toLowerCase()) {
+      return `${prefix}: ${message} (${reason})`;
+    }
+    if (message) {
+      return `${prefix}: ${message}`;
+    }
+    if (reason) {
+      return `${prefix}: ${reason}`;
+    }
+  }
+
+  const fallbackSnippet = normalizeErrorSnippet(input.rawText || '');
+  if (fallbackSnippet) {
+    return `${prefix}: ${fallbackSnippet}`;
+  }
+
+  return prefix;
+}
+
+export function isSub2ApiPlatform(platform?: string | null): boolean {
+  return (platform || '').trim().toLowerCase() === 'sub2api';
+}
+
+export function isManagedSub2ApiTokenDue(
+  tokenExpiresAt?: number,
+  nowMs = Date.now(),
+): boolean {
+  if (!(typeof tokenExpiresAt === 'number' && Number.isFinite(tokenExpiresAt) && tokenExpiresAt > 0)) {
+    return false;
+  }
+  return tokenExpiresAt - nowMs <= SUB2API_MANAGED_REFRESH_LEAD_MS;
+}
+
+export async function refreshSub2ApiManagedSession(params: {
+  account: typeof schema.accounts.$inferSelect;
+  site: typeof schema.sites.$inferSelect;
+  currentAccessToken: string;
+  currentExtraConfig: string | null;
+}): Promise<{ accessToken: string; extraConfig: string }> {
+  const managedAuth = getSub2ApiAuthFromExtraConfig(params.currentExtraConfig);
+  const refreshToken = managedAuth?.refreshToken || '';
+  if (!refreshToken) throw new Error('sub2api managed refresh token missing');
+
+  const endpoint = `${params.site.url.replace(/\/+$/, '')}/api/v1/auth/refresh`;
+  const headers: Record<string, string> = {
+    'Content-Type': 'application/json',
+  };
+  const authHeaderToken = normalizeNonEmptyString(params.currentAccessToken);
+  if (authHeaderToken) {
+    headers.Authorization = `Bearer ${authHeaderToken}`;
+  }
+
+  const { fetch } = await import('undici');
+  let status = 0;
+  let rawText = '';
+  let payload: unknown = null;
+  try {
+    const response = await fetch(endpoint, withSiteRecordProxyRequestInit(params.site, {
+      method: 'POST',
+      headers,
+      body: JSON.stringify({ refresh_token: refreshToken }),
+    }, resolveProxyUrlFromExtraConfig(params.currentExtraConfig)));
+    status = response.status;
+    if (typeof response.text === 'function') {
+      rawText = await response.text().catch(() => '');
+      payload = parseJsonPayload(rawText);
+    } else if (typeof response.json === 'function') {
+      payload = await response.json().catch(() => null);
+      rawText = payload == null ? '' : JSON.stringify(payload);
+    }
+  } catch (err: any) {
+    throw new Error(err?.message || 'sub2api token refresh request failed');
+  }
+
+  const refreshed = parseSub2ApiRefreshPayload(payload);
+  if (!refreshed) {
+    throw new Error(buildSub2ApiRefreshFailureMessage({
+      status,
+      rawText,
+      payload,
+    }));
+  }
+
+  const nextExtraConfig = mergeAccountExtraConfig(params.currentExtraConfig, {
+    sub2apiAuth: {
+      refreshToken: refreshed.refreshToken,
+      tokenExpiresAt: refreshed.tokenExpiresAt,
+    },
+  });
+  await db.update(schema.accounts)
+    .set({
+      accessToken: refreshed.accessToken,
+      extraConfig: nextExtraConfig,
+      status: params.account.status === 'expired' ? 'active' : params.account.status,
+      updatedAt: new Date().toISOString(),
+    })
+    .where(eq(schema.accounts.id, params.account.id))
+    .run();
+
+  return {
+    accessToken: refreshed.accessToken,
+    extraConfig: nextExtraConfig,
+  };
+}
@@ -0,0 +1,290 @@
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
+import { mkdtempSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+
+const refreshSub2ApiManagedSessionSingleflightMock = vi.fn();
+
+vi.mock('./sub2apiRefreshSingleflight.js', () => ({
+  refreshSub2ApiManagedSessionSingleflight: (...args: unknown[]) => refreshSub2ApiManagedSessionSingleflightMock(...args),
+}));
+
+vi.mock('./sub2apiManagedAuth.js', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('./sub2apiManagedAuth.js')>();
+  return {
+    ...actual,
+  };
+});
+
+type DbModule = typeof import('../db/index.js');
+type SchedulerModule = typeof import('./sub2apiRefreshScheduler.js');
+
+function buildSub2ApiExtraConfig(input: {
+  refreshToken?: string;
+  tokenExpiresAt?: number;
+}): string {
+  return JSON.stringify({
+    credentialMode: 'session',
+    ...(input.refreshToken || input.tokenExpiresAt
+      ? {
+          sub2apiAuth: {
+            ...(input.refreshToken ? { refreshToken: input.refreshToken } : {}),
+            ...(input.tokenExpiresAt ? { tokenExpiresAt: input.tokenExpiresAt } : {}),
+          },
+        }
+      : {}),
+  });
+}
+
+describe('sub2apiRefreshScheduler', () => {
+  let db: DbModule['db'];
+  let schema: DbModule['schema'];
+  let executeSub2ApiManagedRefreshPass: SchedulerModule['executeSub2ApiManagedRefreshPass'];
+  let sub2ApiRefreshSchedulerConcurrency: number;
+  let startSub2ApiManagedRefreshScheduler: SchedulerModule['startSub2ApiManagedRefreshScheduler'];
+  let stopSub2ApiManagedRefreshScheduler: SchedulerModule['stopSub2ApiManagedRefreshScheduler'];
+  let resetSub2ApiManagedRefreshSchedulerForTests: SchedulerModule['__resetSub2ApiManagedRefreshSchedulerForTests'];
+  let dataDir = '';
+  let originalDataDir: string | undefined;
+
+  beforeAll(async () => {
+    dataDir = mkdtempSync(join(tmpdir(), 'metapi-sub2api-refresh-scheduler-'));
+    originalDataDir = process.env.DATA_DIR;
+    process.env.DATA_DIR = dataDir;
+
+    await import('../db/migrate.js');
+    const dbModule = await import('../db/index.js');
+    const schedulerModule = await import('./sub2apiRefreshScheduler.js');
+
+    db = dbModule.db;
+    schema = dbModule.schema;
+    executeSub2ApiManagedRefreshPass = schedulerModule.executeSub2ApiManagedRefreshPass;
+    sub2ApiRefreshSchedulerConcurrency = schedulerModule.SUB2API_REFRESH_SCHEDULER_CONCURRENCY;
+    startSub2ApiManagedRefreshScheduler = schedulerModule.startSub2ApiManagedRefreshScheduler;
+    stopSub2ApiManagedRefreshScheduler = schedulerModule.stopSub2ApiManagedRefreshScheduler;
+    resetSub2ApiManagedRefreshSchedulerForTests = schedulerModule.__resetSub2ApiManagedRefreshSchedulerForTests;
+  });
+
+  beforeEach(async () => {
+    vi.useFakeTimers();
+    refreshSub2ApiManagedSessionSingleflightMock.mockReset();
+    await stopSub2ApiManagedRefreshScheduler();
+    await resetSub2ApiManagedRefreshSchedulerForTests();
+
+    await db.delete(schema.accounts).run();
+    await db.delete(schema.sites).run();
+  });
+
+  afterEach(async () => {
+    await stopSub2ApiManagedRefreshScheduler();
+    await resetSub2ApiManagedRefreshSchedulerForTests();
+    vi.useRealTimers();
+  });
+
+  afterAll(() => {
+    if (originalDataDir === undefined) {
+      delete process.env.DATA_DIR;
+    } else {
+      process.env.DATA_DIR = originalDataDir;
+    }
+  });
+
+  it('refreshes only active sub2api session accounts whose managed token is within the refresh lead window', async () => {
+    const nowMs = Date.parse('2026-04-06T02:00:00.000Z');
+    vi.setSystemTime(nowMs);
+
+    const activeSub2ApiSite = await db.insert(schema.sites).values({
+      name: 'sub2-due-site',
+      url: 'https://sub2-due.example.com',
+      platform: 'sub2api',
+      status: 'active',
+    }).returning().get();
+
+    const disabledSub2ApiSite = await db.insert(schema.sites).values({
+      name: 'sub2-disabled-site',
+      url: 'https://sub2-disabled.example.com',
+      platform: 'sub2api',
+      status: 'disabled',
+    }).returning().get();
+
+    const activeNonSub2ApiSite = await db.insert(schema.sites).values({
+      name: 'other-site',
+      url: 'https://other.example.com',
+      platform: 'new-api',
+      status: 'active',
+    }).returning().get();
+
+    const accountIds: Record<string, number> = {};
+    for (const account of [
+      {
+        key: 'due',
+        siteId: activeSub2ApiSite.id,
+        status: 'active',
+        refreshToken: 'sub2-refresh-due',
+        tokenExpiresAt: nowMs + 60_000,
+      },
+      {
+        key: 'expired',
+        siteId: activeSub2ApiSite.id,
+        status: 'active',
+        refreshToken: 'sub2-refresh-expired',
+        tokenExpiresAt: nowMs - 1_000,
+      },
+      {
+        key: 'later',
+        siteId: activeSub2ApiSite.id,
+        status: 'active',
+        refreshToken: 'sub2-refresh-later',
+        tokenExpiresAt: nowMs + (10 * 60 * 1000),
+      },
+      {
+        key: 'inactive',
+        siteId: activeSub2ApiSite.id,
+        status: 'disabled',
+        refreshToken: 'sub2-refresh-inactive',
+        tokenExpiresAt: nowMs + 60_000,
+      },
+      {
+        key: 'disabled_site',
+        siteId: disabledSub2ApiSite.id,
+        status: 'active',
+        refreshToken: 'sub2-refresh-disabled-site',
+        tokenExpiresAt: nowMs + 60_000,
+      },
+      {
+        key: 'missing_refresh_token',
+        siteId: activeSub2ApiSite.id,
+        status: 'active',
+        refreshToken: '',
+        tokenExpiresAt: nowMs + 60_000,
+      },
+      {
+        key: 'missing_expiry',
+        siteId: activeSub2ApiSite.id,
+        status: 'active',
+        refreshToken: 'sub2-refresh-missing-expiry',
+      },
+      {
+        key: 'wrong_platform',
+        siteId: activeNonSub2ApiSite.id,
+        status: 'active',
+        refreshToken: 'sub2-refresh-other-platform',
+        tokenExpiresAt: nowMs + 60_000,
+      },
+    ]) {
+      const inserted = await db.insert(schema.accounts).values({
+        siteId: account.siteId,
+        username: `${account.key}@example.com`,
+        accessToken: `${account.key}-access-token`,
+        apiToken: null,
+        status: account.status,
+        extraConfig: buildSub2ApiExtraConfig({
+          refreshToken: account.refreshToken,
+          tokenExpiresAt: account.tokenExpiresAt,
+        }),
+      }).returning().get();
+      accountIds[account.key] = inserted.id;
+    }
+
+    refreshSub2ApiManagedSessionSingleflightMock.mockResolvedValue({
+      accessToken: 'refreshed-access-token',
+      extraConfig: buildSub2ApiExtraConfig({
+        refreshToken: 'refreshed-refresh-token',
+        tokenExpiresAt: nowMs + (60 * 60 * 1000),
+      }),
+    });
+
+    const result = await executeSub2ApiManagedRefreshPass({ nowMs });
+
+    expect(result).toMatchObject({
+      scanned: 5,
+      refreshed: 2,
+      failed: 0,
+      skipped: 3,
+    });
+    expect(result.refreshedAccountIds.sort((a, b) => a - b)).toEqual([
+      accountIds.due,
+      accountIds.expired,
+    ].sort((a, b) => a - b));
+    expect(result.failedAccountIds).toEqual([]);
+    expect(
+      refreshSub2ApiManagedSessionSingleflightMock.mock.calls
+        .map((call) => call[0]?.account?.id)
+        .sort((a, b) => a - b),
+    ).toEqual([
+      accountIds.due,
+      accountIds.expired,
+    ].sort((a, b) => a - b));
+  });
+
+  it('refreshes due sub2api accounts with bounded concurrency instead of serially', async () => {
+    vi.useRealTimers();
+    const nowMs = Date.parse('2026-04-06T02:00:00.000Z');
+    vi.setSystemTime(nowMs);
+
+    const activeSub2ApiSite = await db.insert(schema.sites).values({
+      name: 'sub2-concurrency-site',
+      url: 'https://sub2-concurrency.example.com',
+      platform: 'sub2api',
+      status: 'active',
+    }).returning().get();
+
+    for (const key of [
+      ...Array.from({ length: sub2ApiRefreshSchedulerConcurrency + 2 }, (_value, index) => `due-${index + 1}`),
+      'later',
+    ]) {
+      await db.insert(schema.accounts).values({
+        siteId: activeSub2ApiSite.id,
+        username: `${key}@example.com`,
+        accessToken: `${key}-access-token`,
+        apiToken: null,
+        status: 'active',
+        extraConfig: buildSub2ApiExtraConfig({
+          refreshToken: `${key}-refresh-token`,
+          tokenExpiresAt: key === 'later' ? nowMs + (10 * 60 * 1000) : nowMs + 60_000,
+        }),
+      }).run();
+    }
+
+    const resolvers: Array<() => void> = [];
+    refreshSub2ApiManagedSessionSingleflightMock.mockImplementation(() => new Promise((resolve) => {
+      resolvers.push(() => resolve({
+        accessToken: 'refreshed-access-token',
+        extraConfig: buildSub2ApiExtraConfig({
+          refreshToken: 'refreshed-refresh-token',
+          tokenExpiresAt: nowMs + (60 * 60 * 1000),
+        }),
+      }));
+    }));
+
+    const passPromise = executeSub2ApiManagedRefreshPass({ nowMs });
+    await new Promise((resolve) => setImmediate(resolve));
+    await new Promise((resolve) => setImmediate(resolve));
+
+    expect(refreshSub2ApiManagedSessionSingleflightMock).toHaveBeenCalledTimes(sub2ApiRefreshSchedulerConcurrency);
+
+    const firstBatchResolvers = [...resolvers];
+    for (const resolve of firstBatchResolvers) {
+      resolve();
+    }
+    await new Promise((resolve) => setImmediate(resolve));
+    await new Promise((resolve) => setImmediate(resolve));
+
+    expect(refreshSub2ApiManagedSessionSingleflightMock).toHaveBeenCalledTimes(
+      sub2ApiRefreshSchedulerConcurrency + 2,
+    );
+
+    const secondBatchResolvers = resolvers.slice(firstBatchResolvers.length);
+    for (const resolve of secondBatchResolvers) {
+      resolve();
+    }
+
+    const result = await passPromise;
+    expect(result).toMatchObject({
+      scanned: sub2ApiRefreshSchedulerConcurrency + 3,
+      refreshed: sub2ApiRefreshSchedulerConcurrency + 2,
+      failed: 0,
+      skipped: 1,
+    });
+  });
+});
@@ -0,0 +1,156 @@
+import { and, eq, sql } from 'drizzle-orm';
+import { db, schema } from '../db/index.js';
+import { getSub2ApiAuthFromExtraConfig } from './accountExtraConfig.js';
+import {
+  isManagedSub2ApiTokenDue,
+  isSub2ApiPlatform,
+} from './sub2apiManagedAuth.js';
+import { refreshSub2ApiManagedSessionSingleflight } from './sub2apiRefreshSingleflight.js';
+
+const ACTIVE_STATUS = 'active';
+const SUB2API_PLATFORM = 'sub2api';
+const SUB2API_REFRESH_SCHEDULER_INTERVAL_MS = 60_000;
+export const SUB2API_REFRESH_SCHEDULER_CONCURRENCY = 4;
+
+let sub2ApiRefreshSchedulerTimer: ReturnType<typeof setInterval> | null = null;
+let sub2ApiRefreshPassInFlight: Promise<void> | null = null;
+
+function clearSub2ApiRefreshSchedulerTimer(): void {
+  if (!sub2ApiRefreshSchedulerTimer) return;
+  clearInterval(sub2ApiRefreshSchedulerTimer);
+  sub2ApiRefreshSchedulerTimer = null;
+}
+
+function normalizeLifecycleStatus(value?: string | null): string {
+  if (typeof value !== 'string') return ACTIVE_STATUS;
+  const normalized = value.trim().toLowerCase();
+  return normalized || ACTIVE_STATUS;
+}
+
+function normalizedLifecycleStatusSql(column: typeof schema.accounts.status | typeof schema.sites.status) {
+  return sql<string>`coalesce(nullif(lower(trim(${column})), ''), ${ACTIVE_STATUS})`;
+}
+
+function normalizedPlatformSql(column: typeof schema.sites.platform) {
+  return sql<string>`coalesce(lower(trim(${column})), '')`;
+}
+
+function shouldRefreshManagedSub2ApiAccount(input: {
+  account: typeof schema.accounts.$inferSelect;
+  site: typeof schema.sites.$inferSelect;
+  nowMs: number;
+}): boolean {
+  if (!isSub2ApiPlatform(input.site.platform)) return false;
+  if (normalizeLifecycleStatus(input.account.status) !== ACTIVE_STATUS) return false;
+  if (normalizeLifecycleStatus(input.site.status) !== ACTIVE_STATUS) return false;
+
+  const managedAuth = getSub2ApiAuthFromExtraConfig(input.account.extraConfig);
+  if (!managedAuth?.refreshToken || !managedAuth.tokenExpiresAt) return false;
+
+  return isManagedSub2ApiTokenDue(managedAuth.tokenExpiresAt, input.nowMs);
+}
+
+export async function executeSub2ApiManagedRefreshPass(input: {
+  nowMs?: number;
+} = {}) {
+  const nowMs = typeof input.nowMs === 'number' && Number.isFinite(input.nowMs)
+    ? input.nowMs
+    : Date.now();
+  const rows = await db.select()
+    .from(schema.accounts)
+    .innerJoin(schema.sites, eq(schema.accounts.siteId, schema.sites.id))
+    .where(and(
+      sql`${normalizedLifecycleStatusSql(schema.accounts.status)} = ${ACTIVE_STATUS}`,
+      sql`${normalizedLifecycleStatusSql(schema.sites.status)} = ${ACTIVE_STATUS}`,
+      sql`${normalizedPlatformSql(schema.sites.platform)} = ${SUB2API_PLATFORM}`,
+    ))
+    .all();
+
+  const refreshCandidates = rows.filter((row) => shouldRefreshManagedSub2ApiAccount({
+    account: row.accounts,
+    site: row.sites,
+    nowMs,
+  }));
+  const refreshedAccountIds: number[] = [];
+  const failedAccountIds: number[] = [];
+  const skipped = rows.length - refreshCandidates.length;
+
+  let cursor = 0;
+  const workerCount = Math.max(1, Math.min(SUB2API_REFRESH_SCHEDULER_CONCURRENCY, refreshCandidates.length || 1));
+  await Promise.all(Array.from({ length: workerCount }, async () => {
+    while (true) {
+      const row = refreshCandidates[cursor];
+      cursor += 1;
+      if (!row) return;
+
+      try {
+        await refreshSub2ApiManagedSessionSingleflight({
+          account: row.accounts,
+          site: row.sites,
+          currentAccessToken: row.accounts.accessToken || '',
+          currentExtraConfig: row.accounts.extraConfig,
+        });
+        refreshedAccountIds.push(row.accounts.id);
+      } catch (error) {
+        failedAccountIds.push(row.accounts.id);
+        console.warn(
+          `[sub2api-refresh] failed to refresh account ${row.accounts.id}: ${(error as Error)?.message || 'unknown error'}`,
+        );
+      }
+    }
+  }));
+
+  return {
+    scanned: rows.length,
+    refreshed: refreshedAccountIds.length,
+    failed: failedAccountIds.length,
+    skipped,
+    refreshedAccountIds,
+    failedAccountIds,
+  };
+}
+
+async function runScheduledSub2ApiRefreshPass(): Promise<void> {
+  if (sub2ApiRefreshPassInFlight) {
+    return sub2ApiRefreshPassInFlight;
+  }
+
+  sub2ApiRefreshPassInFlight = executeSub2ApiManagedRefreshPass()
+    .then(() => undefined)
+    .catch((error) => {
+      console.warn(`[sub2api-refresh] scheduled pass failed: ${(error as Error)?.message || 'unknown error'}`);
+    })
+    .finally(() => {
+      sub2ApiRefreshPassInFlight = null;
+    });
+
+  return sub2ApiRefreshPassInFlight;
+}
+
+export function startSub2ApiManagedRefreshScheduler(intervalMs = SUB2API_REFRESH_SCHEDULER_INTERVAL_MS) {
+  clearSub2ApiRefreshSchedulerTimer();
+
+  const safeIntervalMs = Math.max(SUB2API_REFRESH_SCHEDULER_INTERVAL_MS, Math.trunc(intervalMs || 0));
+  void runScheduledSub2ApiRefreshPass();
+  sub2ApiRefreshSchedulerTimer = setInterval(() => {
+    void runScheduledSub2ApiRefreshPass();
+  }, safeIntervalMs);
+  sub2ApiRefreshSchedulerTimer.unref?.();
+
+  return {
+    enabled: true,
+    intervalMs: safeIntervalMs,
+  };
+}
+
+export async function stopSub2ApiManagedRefreshScheduler() {
+  clearSub2ApiRefreshSchedulerTimer();
+  if (sub2ApiRefreshPassInFlight) {
+    await sub2ApiRefreshPassInFlight;
+  }
+}
+
+export async function __resetSub2ApiManagedRefreshSchedulerForTests() {
+  await stopSub2ApiManagedRefreshScheduler();
+  sub2ApiRefreshPassInFlight = null;
+}
@@ -0,0 +1,99 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+const refreshSub2ApiManagedSessionMock = vi.fn();
+
+vi.mock('./sub2apiManagedAuth.js', () => ({
+  refreshSub2ApiManagedSession: (...args: unknown[]) => refreshSub2ApiManagedSessionMock(...args),
+}));
+
+describe('sub2apiRefreshSingleflight', () => {
+  beforeEach(async () => {
+    refreshSub2ApiManagedSessionMock.mockReset();
+    const { __resetSub2ApiManagedRefreshSingleflightForTests } = await import('./sub2apiRefreshSingleflight.js');
+    __resetSub2ApiManagedRefreshSingleflightForTests();
+  });
+
+  it('coalesces concurrent refreshes for the same account id', async () => {
+    let resolveRefresh: ((value: { accessToken: string; extraConfig: string }) => void) | null = null;
+    refreshSub2ApiManagedSessionMock.mockImplementation(
+      () => new Promise((resolve) => {
+        resolveRefresh = resolve;
+      }),
+    );
+
+    const { refreshSub2ApiManagedSessionSingleflight } = await import('./sub2apiRefreshSingleflight.js');
+
+    const params = {
+      account: { id: 42 } as { id: number },
+      site: { id: 7, platform: 'sub2api', url: 'https://sub2.example.com' } as {
+        id: number;
+        platform: string;
+        url: string;
+      },
+      currentAccessToken: 'stale-access-token',
+      currentExtraConfig: '{"sub2apiAuth":{"refreshToken":"refresh-token"}}',
+    };
+
+    const first = refreshSub2ApiManagedSessionSingleflight(params as never);
+    const second = refreshSub2ApiManagedSessionSingleflight(params as never);
+
+    expect(refreshSub2ApiManagedSessionMock).toHaveBeenCalledTimes(1);
+
+    resolveRefresh?.({
+      accessToken: 'fresh-access-token',
+      extraConfig: '{"sub2apiAuth":{"refreshToken":"next-refresh-token"}}',
+    });
+
+    await expect(first).resolves.toEqual({
+      accessToken: 'fresh-access-token',
+      extraConfig: '{"sub2apiAuth":{"refreshToken":"next-refresh-token"}}',
+    });
+    await expect(second).resolves.toEqual({
+      accessToken: 'fresh-access-token',
+      extraConfig: '{"sub2apiAuth":{"refreshToken":"next-refresh-token"}}',
+    });
+  });
+
+  it('cleans up in-flight state after a rejected refresh so the next attempt can retry', async () => {
+    let rejectRefresh: ((error: Error) => void) | null = null;
+    refreshSub2ApiManagedSessionMock.mockImplementation(
+      () => new Promise((_resolve, reject) => {
+        rejectRefresh = reject;
+      }),
+    );
+
+    const { refreshSub2ApiManagedSessionSingleflight } = await import('./sub2apiRefreshSingleflight.js');
+
+    const params = {
+      account: { id: 42 } as { id: number },
+      site: { id: 7, platform: 'sub2api', url: 'https://sub2.example.com' } as {
+        id: number;
+        platform: string;
+        url: string;
+      },
+      currentAccessToken: 'stale-access-token',
+      currentExtraConfig: '{"sub2apiAuth":{"refreshToken":"refresh-token"}}',
+    };
+
+    const first = refreshSub2ApiManagedSessionSingleflight(params as never);
+    const second = refreshSub2ApiManagedSessionSingleflight(params as never);
+
+    expect(refreshSub2ApiManagedSessionMock).toHaveBeenCalledTimes(1);
+
+    rejectRefresh?.(new Error('refresh rejected'));
+
+    await expect(first).rejects.toThrow('refresh rejected');
+    await expect(second).rejects.toThrow('refresh rejected');
+
+    refreshSub2ApiManagedSessionMock.mockResolvedValueOnce({
+      accessToken: 'fresh-access-token',
+      extraConfig: '{"sub2apiAuth":{"refreshToken":"next-refresh-token"}}',
+    });
+
+    await expect(refreshSub2ApiManagedSessionSingleflight(params as never)).resolves.toEqual({
+      accessToken: 'fresh-access-token',
+      extraConfig: '{"sub2apiAuth":{"refreshToken":"next-refresh-token"}}',
+    });
+    expect(refreshSub2ApiManagedSessionMock).toHaveBeenCalledTimes(2);
+  });
+});
@@ -0,0 +1,28 @@
+import type { schema } from '../db/index.js';
+import { refreshSub2ApiManagedSession } from './sub2apiManagedAuth.js';
+
+type RefreshParams = {
+  account: typeof schema.accounts.$inferSelect;
+  site: typeof schema.sites.$inferSelect;
+  currentAccessToken: string;
+  currentExtraConfig: string | null;
+};
+
+const refreshInFlight = new Map<number, Promise<Awaited<ReturnType<typeof refreshSub2ApiManagedSession>>>>();
+
+export async function refreshSub2ApiManagedSessionSingleflight(params: RefreshParams) {
+  const existing = refreshInFlight.get(params.account.id);
+  if (existing) {
+    return existing;
+  }
+
+  const promise = refreshSub2ApiManagedSession(params).finally(() => {
+    refreshInFlight.delete(params.account.id);
+  });
+  refreshInFlight.set(params.account.id, promise);
+  return promise;
+}
+
+export function __resetSub2ApiManagedRefreshSingleflightForTests() {
+  refreshInFlight.clear();
+}