| 301 | } |
| 302 | |
| 303 | async waitForResult<T>(normalizedKey: string, storageMethod: 'redis' | 'database'): Promise<T> { |
| 304 | const startTime = Date.now() |
| 305 | const redisKey = `${REDIS_KEY_PREFIX}${normalizedKey}` |
| 306 | |
| 307 | while (Date.now() - startTime < MAX_WAIT_TIME_MS) { |
| 308 | let currentResult: ProcessingResult | null = null |
| 309 | |
| 310 | if (storageMethod === 'redis') { |
| 311 | const redis = getRedisClient() |
| 312 | if (!redis) { |
| 313 | throw new Error('Redis not available') |
| 314 | } |
| 315 | const data = await redis.get(redisKey) |
| 316 | currentResult = data ? JSON.parse(data) : null |
| 317 | } else { |
| 318 | const existing = await db |
| 319 | .select({ result: idempotencyKey.result }) |
| 320 | .from(idempotencyKey) |
| 321 | .where(eq(idempotencyKey.key, normalizedKey)) |
| 322 | .limit(1) |
| 323 | currentResult = existing.length > 0 ? (existing[0].result as ProcessingResult) : null |
| 324 | } |
| 325 | |
| 326 | if (currentResult?.status === 'completed') { |
| 327 | logger.debug(`Operation completed, returning result: ${normalizedKey}`) |
| 328 | if (currentResult.success === false) { |
| 329 | throw new Error(currentResult.error || 'Previous operation failed') |
| 330 | } |
| 331 | return currentResult.result as T |
| 332 | } |
| 333 | |
| 334 | if (currentResult?.status === 'failed') { |
| 335 | logger.debug(`Operation failed, throwing error: ${normalizedKey}`) |
| 336 | throw new Error(currentResult.error || 'Previous operation failed') |
| 337 | } |
| 338 | |
| 339 | await sleep(POLL_INTERVAL_MS) |
| 340 | } |
| 341 | |
| 342 | throw new Error(`Timeout waiting for idempotency operation to complete: ${normalizedKey}`) |
| 343 | } |
| 344 | |
| 345 | async storeResult( |
| 346 | normalizedKey: string, |