AI Queue Worker

for (const item of items) {
  const result = await aiClient.process(item);
  await saveResult(result);
}

[데이터 수집] → [Redis Queue] → [AI Worker] → [결과 저장]
                     ↑                |
                     └── 실패 시 재삽입 ──┘

abstract class AbstractQueueWorker<T> {
  protected readonly nameTag: string;
  protected readonly redis: RedisClient;

  constructor(nameTag: string, redis: RedisClient) {
    this.nameTag = nameTag;
    this.redis = redis;
  }

  async start(): Promise<void> {
    console.log(`[${this.nameTag}] 작업 시작`);
    const startTime = Date.now();

    try {
      await this.processQueue();
    } catch (error) {
      console.error(`[${this.nameTag}] 처리 중 오류: ${error.message}`);
    }

    const elapsed = (Date.now() - startTime) / 1000;
    console.log(`[${this.nameTag}] 완료 (${elapsed}s)`);
  }

  // 하위 클래스가 구현할 메서드들
  protected abstract processQueue(): Promise<void>;
  protected abstract getQueueKey(): string;
  protected abstract parseQueueMessage(message: string): T;
  protected abstract processItem(item: T): Promise<void>;
  protected abstract handleFailure(item: T, error: Error): Promise<void>;
}

interface AIQueueItem {
  id: number;
  content: string;
  summary?: string;
  tagList?: string[];
  deathCount: number;  // 실패 횟수 추적
}

class AIWorker extends AbstractQueueWorker<AIQueueItem> {
  private readonly aiClient: AIClient;

  constructor(redis: RedisClient, aiClient: AIClient) {
    super('[AI Service]', redis);
    this.aiClient = aiClient;
  }

  protected async processQueue(): Promise<void> {
    const items = await this.loadFromQueue();
    await Promise.all(items.map((item) => this.processItem(item)));
  }

  protected getQueueKey(): string {
    return 'feed:ai:queue';
  }

  protected parseQueueMessage(message: string): AIQueueItem {
    return JSON.parse(message);
  }

  protected async processItem(item: AIQueueItem): Promise<void> {
    try {
      const result = await this.requestAI(item);
      await this.saveResult(result);
    } catch (error) {
      await this.handleFailure(item, error);
    }
  }

  // ... (아래에서 계속)
}

private async loadFromQueue(): Promise<AIQueueItem[]> {
  const BATCH_SIZE = parseInt(process.env.AI_RATE_LIMIT_COUNT);

  const results = await this.redis.executePipeline((pipeline) => {
    for (let i = 0; i < BATCH_SIZE; i++) {
      pipeline.rpop(this.getQueueKey());
    }
  });

  return results
    .map(([err, value]) => JSON.parse(value as string))
    .filter((item) => item !== null);
}

private isRetryableError(error: Error): boolean {
  const message = error.message.toLowerCase();

  // 영구적 에러 → 재시도 안 함
  if (message.includes('invalid') || message.includes('401')) return false;
  if (message.includes('json') || message.includes('parse')) return false;

  // 일시적 에러 → 재시도
  if (message.includes('rate limit') || message.includes('429')) return true;
  if (message.includes('timeout') || message.includes('503')) return true;

  // 알 수 없는 에러는 일단 재시도 (보수적 접근)
  return true;
}

protected async handleFailure(
  item: AIQueueItem,
  error: Error,
): Promise<void> {
  const shouldRetry = this.isRetryableError(error);

  console.error(
    `[${this.nameTag}] ${item.id} 처리 실패:`,
    `에러: ${error.name} - ${error.message}`,
    `재시도 가능: ${shouldRetry}`,
    `현재 deathCount: ${item.deathCount}`,
  );

  if (shouldRetry && item.deathCount < 3) {
    // 재시도: deathCount 증가시키고 큐 뒤에 다시 삽입
    item.deathCount++;
    await this.redis.rpush(this.getQueueKey(), [JSON.stringify(item)]);
    console.warn(`[${this.nameTag}] ${item.id} 재시도 예약 (${item.deathCount}/3)`);
  } else {
    // 영구 실패: 기본값으로 저장하고 넘어감
    const reason = shouldRetry
      ? 'Death Count 3회 초과'
      : `재시도 불가능한 에러 (${error.name})`;
    console.error(`[${this.nameTag}] ${item.id} 영구 실패 - ${reason}`);
    await this.saveDefaultResult(item.id);
  }
}

에러 발생
  │
  ├─ 영구적 에러? ──→ 즉시 포기 → 기본값 저장
  │
  └─ 일시적 에러?
       │
       ├─ deathCount < 3? ──→ deathCount++ → 큐에 재삽입
       │
       └─ deathCount >= 3? ──→ 포기 → 기본값 저장

private async requestAI(item: AIQueueItem): Promise<AIQueueItem> {
  const params = {
    max_tokens: 8192,
    system: SYSTEM_PROMPT,
    messages: [{ role: 'user', content: item.content }],
    model: 'claude-3-5-haiku-latest',
  };

  const response = await this.aiClient.messages.create(params);
  const text = response.content[0].text.replace(/[\n\r\t\s]+/g, ' ');
  const parsed = JSON.parse(text);

  item.summary = parsed.summary;
  item.tagList = Object.keys(parsed.tags);

  return item;
}

private async saveResult(item: AIQueueItem): Promise<void> {
  // DB에 태그 관계 저장
  await this.tagRepository.insertTags(item.id, item.tagList);

  // Redis 캐시에 태그 저장 (빠른 조회용)
  await this.redis.hset(`feed:recent:${item.id}`, 'tag', item.tagList.join(','));

  // DB에 요약 저장
  await this.feedRepository.updateSummary(item.id, item.summary);
}

import * as schedule from 'node-schedule';

// 매 분마다 워커 실행
schedule.scheduleJob('*/1 * * * *', async () => {
  await aiWorker.start();
});

const queueLength = await redis.llen('feed:ai:queue');
if (queueLength > THRESHOLD) {
  // 알림 발송 또는 배치 크기 증가
}