File size: 6,693 Bytes
9e4583c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
/**
 * Proactive Token Health Check Scheduler
 *
 * Background job that periodically refreshes OAuth tokens before they expire.
 * Each connection can configure its own `healthCheckInterval` (minutes).
 * Default: 60 minutes.  0 = disabled.
 *
 * The scheduler runs a lightweight sweep every TICK_MS (60 s).
 * For each eligible connection it calls the provider-specific refresh function,
 * updates the DB, and logs the result.
 */

import { getProviderConnections, updateProviderConnection } from "@/lib/localDb";
import {
  getAccessToken,
  supportsTokenRefresh,
  isUnrecoverableRefreshError,
} from "@omniroute/open-sse/services/tokenRefresh.ts";

// ── Constants ────────────────────────────────────────────────────────────────
const TICK_MS = 60 * 1000; // sweep interval: every 60 seconds
const DEFAULT_HEALTH_CHECK_INTERVAL_MIN = 60; // default per-connection interval
const LOG_PREFIX = "[HealthCheck]";

// ── Singleton guard ──────────────────────────────────────────────────────────
let initialized = false;
let intervalHandle = null;

/**
 * Start the health-check scheduler (idempotent).
 */
export function initTokenHealthCheck() {
  if (initialized) return;
  initialized = true;

  console.log(
    `${LOG_PREFIX} Starting proactive token health-check (tick every ${TICK_MS / 1000}s)`
  );

  // Run first sweep after a short delay so the server finishes booting
  setTimeout(() => {
    sweep();
    intervalHandle = setInterval(sweep, TICK_MS);
  }, 10_000);
}

/**
 * Stop the scheduler (useful for tests / hot-reload).
 */
export function stopTokenHealthCheck() {
  if (intervalHandle) {
    clearInterval(intervalHandle);
    intervalHandle = null;
  }
  initialized = false;
}

// ── Core sweep ───────────────────────────────────────────────────────────────
async function sweep() {
  try {
    const connections = await getProviderConnections({ authType: "oauth" });

    if (!connections || connections.length === 0) return;

    for (const conn of connections) {
      try {
        await checkConnection(conn);
      } catch (err) {
        // Per-connection isolation: one failure never blocks others
        console.error(`${LOG_PREFIX} Error checking ${conn.name || conn.id}:`, err.message);
      }
    }
  } catch (err) {
    console.error(`${LOG_PREFIX} Sweep error:`, err.message);
  }
}

/**
 * Check a single connection and refresh if due.
 */
async function checkConnection(conn) {
  // Determine interval (0 = disabled)
  const intervalMin = conn.healthCheckInterval ?? DEFAULT_HEALTH_CHECK_INTERVAL_MIN;
  if (intervalMin <= 0) return;
  if (!conn.isActive) return;
  if (!conn.refreshToken || typeof conn.refreshToken !== "string") return;

  // Skip connections already marked as expired (need re-auth, not retry)
  if (conn.testStatus === "expired") return;

  if (!supportsTokenRefresh(conn.provider)) {
    const now = new Date().toISOString();
    await updateProviderConnection(conn.id, { lastHealthCheckAt: now });
    console.log(
      `${LOG_PREFIX} Skipping ${conn.provider}/${conn.name || conn.email || conn.id} (refresh unsupported)`
    );
    return;
  }

  const intervalMs = intervalMin * 60 * 1000;
  const lastCheck = conn.lastHealthCheckAt ? new Date(conn.lastHealthCheckAt).getTime() : 0;

  // Not yet due
  if (Date.now() - lastCheck < intervalMs) return;

  console.log(
    `${LOG_PREFIX} Refreshing ${conn.provider}/${conn.name || conn.email || conn.id} (interval: ${intervalMin}min)`
  );

  const credentials = {
    refreshToken: conn.refreshToken,
    accessToken: conn.accessToken,
    expiresAt: conn.tokenExpiresAt,
    providerSpecificData: conn.providerSpecificData,
  };

  const result = await getAccessToken(conn.provider, credentials, {
    info: (tag, msg) => console.log(`${LOG_PREFIX} [${tag}] ${msg}`),
    warn: (tag, msg) => console.warn(`${LOG_PREFIX} [${tag}] ${msg}`),
    error: (tag, msg, extra) => console.error(`${LOG_PREFIX} [${tag}] ${msg}`, extra || ""),
  });

  const now = new Date().toISOString();

  // ─── Handle unrecoverable errors (e.g. refresh_token_reused) ───────────
  // OpenAI Codex uses rotating one-time-use refresh tokens.
  // Once used, the old token is permanently invalidated.
  // Retrying will never succeed β†’ deactivate and stop the loop.
  if (isUnrecoverableRefreshError(result)) {
    await updateProviderConnection(conn.id, {
      lastHealthCheckAt: now,
      testStatus: "expired",
      lastError: `Refresh token consumed (${result.error}). Please re-authenticate this account.`,
      lastErrorAt: now,
      lastErrorType: result.error,
      lastErrorSource: "oauth",
      errorCode: result.error,
      isActive: false,
      refreshToken: null,
    });
    console.error(
      `${LOG_PREFIX} βœ— ${conn.provider}/${conn.name || conn.email || conn.id} β€” ` +
        `Refresh token is permanently invalid (${result.error}). ` +
        `Connection deactivated. Re-authenticate to restore.`
    );
    return;
  }

  if (result && result.accessToken) {
    // Token refreshed successfully β€” update DB
    const updateData: any = {
      accessToken: result.accessToken,
      lastHealthCheckAt: now,
      testStatus: "active",
      lastError: null,
      lastErrorAt: null,
      lastErrorType: null,
      lastErrorSource: null,
      errorCode: null,
    };

    if (result.refreshToken) {
      updateData.refreshToken = result.refreshToken;
    }

    if (result.expiresIn) {
      updateData.tokenExpiresAt = new Date(Date.now() + result.expiresIn * 1000).toISOString();
    }

    await updateProviderConnection(conn.id, updateData);
    console.log(`${LOG_PREFIX} βœ“ ${conn.provider}/${conn.name || conn.email || conn.id} refreshed`);
  } else {
    // Refresh failed β€” record but don't disable the connection
    await updateProviderConnection(conn.id, {
      lastHealthCheckAt: now,
      testStatus: "error",
      lastError: "Health check: token refresh failed",
      lastErrorAt: now,
      lastErrorType: "token_refresh_failed",
      lastErrorSource: "oauth",
      errorCode: "refresh_failed",
    });
    console.warn(
      `${LOG_PREFIX} βœ— ${conn.provider}/${conn.name || conn.email || conn.id} refresh failed`
    );
  }
}

// Auto-start when imported
initTokenHealthCheck();

export default initTokenHealthCheck;