feat: add batched recipe import queue

2026-08-28 11:09:36 +09:00
parent 4491c91455
commit f65e660184
19 changed files with 1080 additions and 262 deletions
@@ -7,20 +7,29 @@ import {
  createTestConfig,
  FakeAllowedEmailRepository,
  FakeGroupRepository,
+  FakeImportJobRepository,
  FakeRecipeRepository,
  fakeUserRepository,
  recipeDraft,
  source,
 } from './helpers/fakes.js';

-function createDependencies(recipeRepository, groupRepository = new FakeGroupRepository()) {
+function createDependencies(
+  recipeRepository,
+  groupRepository = new FakeGroupRepository(),
+  importJobRepository = new FakeImportJobRepository(),
+) {
  return {
    allowedEmailRepository: new FakeAllowedEmailRepository(),
    groupRepository,
+    importJobRepository,
    recipeRepository,
    userRepository: fakeUserRepository,
    sourceExtractor: { extract: async () => source },
-    recipeParser: { parse: async () => recipeDraft },
+    recipeParser: {
+      parse: async () => recipeDraft,
+      parseMany: async (entries) => entries.map(({ jobId }) => ({ jobId, recipe: recipeDraft })),
+    },
    imageStorage: {
      saveFromUrl: async (id) => `recipes/${id}/cover.webp`,
      removeRecipe: async () => {},
@@ -74,6 +83,10 @@ test('관리자는 이메일만으로 Google 로그인 허용 계정을 관리
  assert.match(pageResponse.body, /id="site-header"/);
  assert.match(pageResponse.body, /id="access-overlay"/);
  assert.match(pageResponse.body, /id="scroll-to-top"/);
+  assert.match(pageResponse.body, /id="source-urls"/);
+  assert.match(pageResponse.body, /id="import-queue"/);
+  assert.match(pageResponse.body, /id="process-import-jobs"/);
+  assert.doesNotMatch(pageResponse.body, /id="preview-section"/);

  const appBarResponse = await app.inject({ method: 'GET', url: '/js/app-bar.js' });
  assert.equal(appBarResponse.statusCode, 200);
@@ -339,6 +352,103 @@ test('import preview와 소유자별 Recipe CRUD가 이어진다', async (t) =>
  assert.equal(deleteResponse.statusCode, 204);
 });

+test('여러 URL을 대기열에 넣고 비동기로 일괄 처리하며 완료 작업은 숨긴다', async (t) => {
+  const recipeRepository = new FakeRecipeRepository();
+  const importJobRepository = new FakeImportJobRepository();
+  const dependencies = createDependencies(
+    recipeRepository,
+    new FakeGroupRepository(),
+    importJobRepository,
+  );
+  const app = await buildApp({ config: createTestConfig(), dependencies });
+  t.after(() => app.close());
+  const cookie = authCookie(app);
+
+  const enqueueResponse = await app.inject({
+    method: 'POST',
+    url: '/api/import/jobs',
+    headers: { cookie },
+    payload: { urls: [source.sourceUrl, source.sourceUrl] },
+  });
+  assert.equal(enqueueResponse.statusCode, 201);
+  assert.equal(enqueueResponse.json().addedCount, 1);
+  assert.equal(enqueueResponse.json().duplicateCount, 1);
+  assert.equal(enqueueResponse.json().jobs[0].status, 'pending');
+
+  const processResponse = await app.inject({
+    method: 'POST',
+    url: '/api/import/jobs/process',
+    headers: { cookie },
+  });
+  assert.equal(processResponse.statusCode, 202);
+  await app.importJobProcessor.waitForIdle('google-user-1');
+
+  const queueResponse = await app.inject({
+    method: 'GET',
+    url: '/api/import/jobs',
+    headers: { cookie },
+  });
+  assert.deepEqual(queueResponse.json().jobs, []);
+
+  const recipesResponse = await app.inject({
+    method: 'GET',
+    url: '/api/recipes',
+    headers: { cookie },
+  });
+  assert.equal(recipesResponse.json().recipes.length, 1);
+
+  const duplicateResponse = await app.inject({
+    method: 'POST',
+    url: '/api/import/jobs',
+    headers: { cookie },
+    payload: { urls: [source.sourceUrl] },
+  });
+  assert.equal(duplicateResponse.json().addedCount, 0);
+  assert.equal(duplicateResponse.json().duplicateCount, 1);
+});
+
+test('가져오기 실패 작업은 대기열에 남고 사용자가 재시도할 수 있다', async (t) => {
+  const dependencies = createDependencies(new FakeRecipeRepository());
+  dependencies.sourceExtractor = {
+    async extract() {
+      throw new Error('원문 추출 실패');
+    },
+  };
+  const app = await buildApp({ config: createTestConfig(), dependencies });
+  t.after(() => app.close());
+  const cookie = authCookie(app);
+
+  await app.inject({
+    method: 'POST',
+    url: '/api/import/jobs',
+    headers: { cookie },
+    payload: { urls: [source.sourceUrl] },
+  });
+  await app.inject({
+    method: 'POST',
+    url: '/api/import/jobs/process',
+    headers: { cookie },
+  });
+  await app.importJobProcessor.waitForIdle('google-user-1');
+
+  const failedResponse = await app.inject({
+    method: 'GET',
+    url: '/api/import/jobs',
+    headers: { cookie },
+  });
+  const [failedJob] = failedResponse.json().jobs;
+  assert.equal(failedJob.status, 'failed');
+  assert.equal(failedJob.error.message, '가져오기 작업을 처리하지 못했습니다.');
+
+  const retryResponse = await app.inject({
+    method: 'POST',
+    url: `/api/import/jobs/${failedJob.id}/retry`,
+    headers: { cookie },
+  });
+  assert.equal(retryResponse.statusCode, 200);
+  assert.equal(retryResponse.json().job.status, 'pending');
+});
+
 test('같은 그룹 사용자는 레시피를 함께 조회하지만 작성자만 수정하고 삭제한다', async (t) => {
  const recipeRepository = new FakeRecipeRepository();
  const groupRepository = new FakeGroupRepository([
@@ -60,6 +60,107 @@ export class FakeRecipeRepository {
  }
 }

+export class FakeImportJobRepository {
+  constructor() {
+    this.jobs = new Map();
+    this.nextId = 1;
+  }
+
+  async enqueue({ ownerGoogleSub, url, platform, sourceId }) {
+    const existing = [...this.jobs.values()].find((job) => (
+      job.ownerGoogleSub === ownerGoogleSub
+      && job.platform === platform
+      && job.sourceId === sourceId
+    ));
+    if (existing) return { created: false, job: existing };
+
+    const now = new Date();
+    const job = {
+      _id: `job-${this.nextId}`,
+      ownerGoogleSub,
+      url,
+      platform,
+      sourceId,
+      status: 'pending',
+      attempts: 0,
+      error: null,
+      recipeId: null,
+      createdAt: now,
+      updatedAt: now,
+      startedAt: null,
+      completedAt: null,
+      lockedAt: null,
+    };
+    this.nextId += 1;
+    this.jobs.set(job._id, job);
+    return { created: true, job };
+  }
+
+  async listVisible(ownerGoogleSub) {
+    return [...this.jobs.values()].filter((job) => (
+      job.ownerGoogleSub === ownerGoogleSub
+      && ['pending', 'processing', 'failed'].includes(job.status)
+    ));
+  }
+
+  async recoverStale(ownerGoogleSub, staleBefore) {
+    for (const job of this.jobs.values()) {
+      if (
+        job.ownerGoogleSub === ownerGoogleSub
+        && job.status === 'processing'
+        && job.lockedAt < staleBefore
+      ) {
+        Object.assign(job, { status: 'pending', startedAt: null, lockedAt: null });
+      }
+    }
+  }
+
+  async claimNext(ownerGoogleSub) {
+    const job = [...this.jobs.values()].find((candidate) => (
+      candidate.ownerGoogleSub === ownerGoogleSub && candidate.status === 'pending'
+    ));
+    if (!job) return null;
+    const now = new Date();
+    Object.assign(job, {
+      status: 'processing',
+      error: null,
+      attempts: job.attempts + 1,
+      startedAt: now,
+      lockedAt: now,
+      updatedAt: now,
+    });
+    return job;
+  }
+
+  async markCompleted(jobId, recipeId) {
+    const job = this.jobs.get(jobId);
+    Object.assign(job, {
+      status: 'completed',
+      recipeId,
+      error: null,
+      completedAt: new Date(),
+      lockedAt: null,
+    });
+  }
+
+  async markFailed(jobId, error) {
+    Object.assign(this.jobs.get(jobId), { status: 'failed', error, lockedAt: null });
+  }
+
+  async retry(ownerGoogleSub, jobId) {
+    const job = this.jobs.get(jobId);
+    if (job?.ownerGoogleSub !== ownerGoogleSub || job.status !== 'failed') return null;
+    Object.assign(job, {
+      status: 'pending',
+      error: null,
+      startedAt: null,
+      lockedAt: null,
+      updatedAt: new Date(),
+    });
+    return job;
+  }
+}
+
 export class FakeGroupRepository {
  constructor(groups = []) {
    this.groups = groups;
@@ -0,0 +1,76 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import {
+  createImportBatches,
+  ImportJobProcessorService,
+} from '../src/services/import-job-processor.service.js';
+import { FakeImportJobRepository, recipeDraft, source } from './helpers/fakes.js';
+
+function extractedItem(jobId, rawText) {
+  return {
+    jobId,
+    source: { rawText },
+  };
+}
+
+test('가져오기 본문은 최대 3개와 합계 100,000자 단위로 묶는다', () => {
+  const batches = createImportBatches([
+    extractedItem('job-1', 'a'.repeat(40_000)),
+    extractedItem('job-2', 'b'.repeat(40_000)),
+    extractedItem('job-3', 'c'.repeat(30_000)),
+    extractedItem('job-4', 'd'),
+    extractedItem('job-5', 'e'),
+  ]);
+
+  assert.deepEqual(batches.map((batch) => batch.map(({ jobId }) => jobId)), [
+    ['job-1', 'job-2'],
+    ['job-3', 'job-4', 'job-5'],
+  ]);
+});
+
+test('처리 요청 하나가 대기열 전체를 3개씩 MiniMax 배치 처리한다', async () => {
+  const importJobRepository = new FakeImportJobRepository();
+  for (let index = 1; index <= 4; index += 1) {
+    await importJobRepository.enqueue({
+      ownerGoogleSub: 'google-user-1',
+      url: `https://www.instagram.com/reel/Recipe${index}/`,
+      platform: 'instagram',
+      sourceId: `Recipe${index}`,
+    });
+  }
+
+  const batchJobIds = [];
+  const processor = new ImportJobProcessorService({
+    importJobRepository,
+    sourceExtractor: {
+      async extract(url) {
+        const sourceId = new URL(url).pathname.split('/').filter(Boolean)[1];
+        return {
+          ...source,
+          sourceUrl: url,
+          sourceId,
+          rawText: `${sourceId} 본문`,
+        };
+      },
+    },
+    recipeParser: {
+      async parseMany(entries) {
+        batchJobIds.push(entries.map(({ jobId }) => jobId));
+        return entries.map(({ jobId }) => ({ jobId, recipe: recipeDraft }));
+      },
+    },
+    recipeCreator: {
+      async create({ source: recipeSource }) {
+        return { _id: `recipe-${recipeSource.sourceId}` };
+      },
+    },
+    logger: { error() {} },
+  });
+
+  processor.start('google-user-1');
+  await processor.waitForIdle('google-user-1');
+
+  assert.deepEqual(batchJobIds.map((ids) => ids.length), [3, 1]);
+  assert.deepEqual(await importJobRepository.listVisible('google-user-1'), []);
+  assert.ok([...importJobRepository.jobs.values()].every(({ status }) => status === 'completed'));
+});
@@ -24,6 +24,7 @@ test('Windows에서 개발용 MongoDB와 로컬 이미지 프로필을 자동
  assert.deepEqual(config.mongoCollections, {
    allowedEmails: 'allowed_google_emails',
    groups: 'groups_dev',
+    importJobs: 'import_jobs_dev',
    recipes: 'recipes_dev',
    users: 'users_dev',
  });
@@ -45,6 +46,7 @@ test('OS 프로필이 환경파일의 개발·운영 선택값보다 우선한
  assert.deepEqual(linuxConfig.mongoCollections, {
    allowedEmails: 'allowed_google_emails',
    groups: 'groups',
+    importJobs: 'import_jobs',
    recipes: 'recipes',
    users: 'users',
  });
@@ -75,3 +75,36 @@ test('MiniMax 사고 과정 분리와 충분한 출력 한도를 요청한다',
  assert.match(request.messages[0].content, /수량, 온도, 시간 값은 변환하지 않습니다/);
  assert.equal(result.ingredientGroups[0].name, '재료');
 });
+
+test('여러 본문과 jobId를 MiniMax 단일 요청으로 보내고 각각 검증한다', async () => {
+  const requests = [];
+  const client = {
+    chat: {
+      completions: {
+        async create(parameters) {
+          requests.push(parameters);
+          const sources = JSON.parse(parameters.messages[1].content).sources;
+          return {
+            choices: [{
+              message: {
+                content: JSON.stringify({
+                  results: sources.map(({ jobId }) => ({ jobId, recipe: recipeDraft })),
+                }),
+              },
+            }],
+          };
+        },
+      },
+    },
+  };
+  const service = new RecipeParserService({ client, model: 'MiniMax-M2.7' });
+  const results = await service.parseMany([
+    { jobId: 'job-1', source: { platform: 'instagram', title: null, rawText: '본문 1' } },
+    { jobId: 'job-2', source: { platform: 'youtube', title: '제목', rawText: '본문 2' } },
+  ]);
+
+  assert.equal(requests.length, 1);
+  assert.equal(requests[0].max_completion_tokens, 16384);
+  assert.deepEqual(results.map(({ jobId }) => jobId), ['job-1', 'job-2']);
+  assert.deepEqual(results.map(({ recipe }) => recipe), [recipeDraft, recipeDraft]);
+});