140 lines
3.9 KiB
TypeScript
140 lines
3.9 KiB
TypeScript
/* eslint-disable no-console, @typescript-eslint/no-explicit-any */
|
|
|
|
import { createClient, RedisClientType } from 'redis';
|
|
|
|
// 连接配置接口
|
|
export interface RedisConnectionConfig {
|
|
url: string;
|
|
clientName: string; // 用于日志显示，如 "Redis" 或 "Pika"
|
|
}
|
|
|
|
// 添加 Redis 操作重试包装器。这个文件只给 Node Redis/Kvrocks 使用，
|
|
// 避免 Cloudflare/D1/Upstash 构建因为 BaseRedisStorage 引入 node-redis。
|
|
export function createRetryWrapper(
|
|
clientName: string,
|
|
getClient: () => RedisClientType
|
|
) {
|
|
return async function withRetry<T>(
|
|
operation: () => Promise<T>,
|
|
maxRetries = 3
|
|
): Promise<T> {
|
|
for (let i = 0; i < maxRetries; i++) {
|
|
try {
|
|
return await operation();
|
|
} catch (err: any) {
|
|
const isLastAttempt = i === maxRetries - 1;
|
|
const isConnectionError =
|
|
err.message?.includes('Connection') ||
|
|
err.message?.includes('ECONNREFUSED') ||
|
|
err.message?.includes('ENOTFOUND') ||
|
|
err.code === 'ECONNRESET' ||
|
|
err.code === 'EPIPE';
|
|
|
|
if (isConnectionError && !isLastAttempt) {
|
|
console.log(
|
|
`${clientName} operation failed, retrying... (${
|
|
i + 1
|
|
}/${maxRetries})`
|
|
);
|
|
console.error('Error:', err.message);
|
|
|
|
// 等待一段时间后重试
|
|
await new Promise((resolve) => setTimeout(resolve, 1000 * (i + 1)));
|
|
|
|
// 尝试重新连接
|
|
try {
|
|
const client = getClient();
|
|
if (!client.isOpen) {
|
|
await client.connect();
|
|
}
|
|
} catch (reconnectErr) {
|
|
console.error('Failed to reconnect:', reconnectErr);
|
|
}
|
|
|
|
continue;
|
|
}
|
|
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
throw new Error('Max retries exceeded');
|
|
};
|
|
}
|
|
|
|
// 创建客户端的工厂函数
|
|
export function createRedisClient(
|
|
config: RedisConnectionConfig,
|
|
globalSymbol: symbol
|
|
): RedisClientType {
|
|
let client: RedisClientType | undefined = (global as any)[globalSymbol];
|
|
|
|
if (!client) {
|
|
if (!config.url) {
|
|
throw new Error(`${config.clientName}_URL env variable not set`);
|
|
}
|
|
|
|
// 创建客户端配置
|
|
const clientConfig: any = {
|
|
url: config.url,
|
|
socket: {
|
|
// 重连策略：指数退避，最大30秒
|
|
reconnectStrategy: (retries: number) => {
|
|
console.log(
|
|
`${config.clientName} reconnection attempt ${retries + 1}`
|
|
);
|
|
if (retries > 10) {
|
|
console.error(
|
|
`${config.clientName} max reconnection attempts exceeded`
|
|
);
|
|
return false; // 停止重连
|
|
}
|
|
return Math.min(1000 * Math.pow(2, retries), 30000); // 指数退避，最大30秒
|
|
},
|
|
connectTimeout: 10000, // 10秒连接超时
|
|
// 设置 no delay，减少延迟
|
|
noDelay: true,
|
|
},
|
|
// 添加其他配置
|
|
pingInterval: 30000, // 30秒 ping 一次，保持连接活跃
|
|
};
|
|
|
|
client = createClient(clientConfig);
|
|
|
|
// 添加错误事件监听
|
|
client.on('error', (err) => {
|
|
console.error(`${config.clientName} client error:`, err);
|
|
});
|
|
|
|
client.on('connect', () => {
|
|
console.log(`${config.clientName} connected`);
|
|
});
|
|
|
|
client.on('reconnecting', () => {
|
|
console.log(`${config.clientName} reconnecting...`);
|
|
});
|
|
|
|
client.on('ready', () => {
|
|
console.log(`${config.clientName} ready`);
|
|
});
|
|
|
|
// 初始连接，带重试机制
|
|
const connectWithRetry = async () => {
|
|
try {
|
|
await client!.connect();
|
|
console.log(`${config.clientName} connected successfully`);
|
|
} catch (err) {
|
|
console.error(`${config.clientName} initial connection failed:`, err);
|
|
console.log('Will retry in 5 seconds...');
|
|
setTimeout(connectWithRetry, 5000);
|
|
}
|
|
};
|
|
|
|
connectWithRetry();
|
|
|
|
(global as any)[globalSymbol] = client;
|
|
}
|
|
|
|
return client;
|
|
}
|