# Queue System BullMQ on Redis. The API process enqueues jobs; the worker process (`src/worker.ts`) consumes them. ## Critical Redis Requirement Redis **must** run with `--maxmemory-policy noeviction`. BullMQ silently loses jobs if Redis uses `allkeys-lru`. This is pre-configured in `infrastructure/docker-compose.yml`. ## Queue Names Defined in `backend/src/queues/queues.constants.ts`: | Constant | Queue name | |---|---| | `QUEUES.YOUTUBE_SYNC` | `youtube-sync` | | `QUEUES.RENDER` | `render` | | `QUEUES.LINT` | `lint` | | `QUEUES.BULK_METADATA` | `bulk-metadata` | | `QUEUES.IMPORT` | `import` | | `QUEUES.CONFLICT_DETECTION` | `conflict-detection` | ## Processors ### `youtube-sync.processor.ts` **Trigger:** `POST /videos/:id/sync` **Job data:** `{ videoId }` **Behavior:** 1. Calls `VideoRenderService` to render description and compute hash 2. Skips push if hash matches `lastSyncedHash` (no changes) 3. Calls `QuotaService.canSpend(50)` — rejects if quota exceeded 4. Pushes all editable fields to YouTube API (`videos.update`) 5. Updates `renderedDescription`, `lastSyncedHash`, `lastSyncedAt` on the video row ### `render.processor.ts` **Trigger:** `PUT /video-configs/:videoId` when the request body includes `"autoRender": true` **Job data:** `{ videoId }` **Behavior:** Renders description without pushing to YouTube. Updates `renderedDescription`. The render job is not enqueued automatically on every config save — it is opt-in via the `autoRender` flag. `VideoConfigsService.upsert()` checks `if (dto.autoRender)` and only then calls `renderQueue.add(...)`. The frontend passes `autoRender: true` when saving from the video editor so the preview updates in the background. Saves that don't need an immediate re-render (e.g. bulk template applies) omit the flag and skip the queue. ### `lint.processor.ts` **Trigger:** `POST /lint/bulk`, `POST /lint/channel/:id`, `POST /lint/team` **Job data:** `{ videoId }` **Behavior:** Calls `LintingService.lintVideo(videoId)`. Replaces all unresolved `LintResult` rows and recomputes `Video.lintStatus`. > **Note:** `POST /lint/videos/:id` (single video) is **synchronous** — it calls `LintingService.lintVideo()` directly in the HTTP handler and returns results immediately. It does not use this queue. Only bulk operations go through BullMQ. ### `bulk-metadata.processor.ts` **Trigger:** Bulk job confirmed by user **Job data:** `{ bulkJobId }` **Behavior:** Processes each `BulkJobItem` in sequence. Updates video fields, saves rollback snapshots, updates job status counters. ### `import.processor.ts` **Trigger:** CSV import committed (`POST /imports/csv/commit`) **Job data:** `{ importJobId }` **Behavior:** Calls `ImportsService.executeCommit()`, which marks the `ImportJob` as `committed` and writes an audit log entry. **Does not create or update any Video rows.** The preview step does not persist the validated rows, so the processor has no data to act on. CSV import is structurally incomplete — see the backlog. ### `conflict-detection.processor.ts` **Trigger:** BullMQ repeatable job registered on worker startup by `ConflictDetectionScheduler`. Cron pattern comes from `CONFLICT_DETECTION_CRON` (default `0 3 * * *`). Registration is gated by the global `CONFLICT_DETECTION_ENABLED` env var. **Job data:** `{}` **Behavior:** 1. Loads all teams with `conflictDetectionEnabled: true` 2. For each team, selects videos where `remoteConflict = false`, `youtubeDeletedAt = null`, and `lastSyncedAt < now - minAgeDays`, stalest-first, capped at `batchSize` 3. Hands the selected video IDs to `YouTubeSyncService.detectConflictsForVideos()`, which groups by `channelId` (needed for OAuth) and issues one `videos.list` call per batch of up to 50 IDs. Cost is **1 quota unit per batch**, not per video (YouTube quota is per method-call, invariant to the number of parts or IDs). `QuotaService.canSpend(1)` is checked before each batch. 4. Per video within a batch: on hash mismatch, writes `pendingRemoteSnapshot` + `pendingRemoteDescription` and sets `remoteConflict = true`. On hash match with a stale flag, clears the flag and pending fields — self-heals when the creator reverts an out-of-band edit. 5. Stops the entire sweep as soon as quota is exhausted; resumes on the next cron tick. `Team.conflictDetectionBatchSize` (1–500) and `conflictDetectionMinAgeDays` (≥0) are per-team knobs. Because the API cost is per batch (up to 50 videos each), a `batchSize` of 500 consumes ~10 units per team per run — not 500. Users resolve detected conflicts via `POST /videos/:id/accept-remote` (adopt remote, zero extra quota) or `POST /videos/:id/sync` (push local, overwrite remote). ## Frontend Job Completion Detection There is **no push mechanism** (no SSE, no WebSocket, no EventEmitter). The frontend detects job completion entirely through polling via TanStack Query's `refetchInterval`. ### Sync queue status — dynamic polling The `
` component (`components/shared/Header.tsx`) polls `GET /youtube-sync/queue-status` continuously with an interval that adapts to queue state: | Queue state | Poll interval | |---|---| | Active jobs running | 3 s | | Jobs waiting | 8 s | | Idle | 30 s | This drives the sync-in-progress indicator in the header. When the queue drains, the indicator clears within one polling cycle. ### Video editor — scheduled invalidations After `POST /videos/:id/sync` is enqueued, the video editor schedules two forced refetches of `['video', id]` — at **3 s** and **8 s** — to pick up the updated `hasPendingChanges` once the processor finishes. If the job takes longer than 8 s, the UI shows stale "push pending" status until the next sync-status poll triggers a broader refresh. ### Bulk jobs page — fixed polling `/bulk-jobs` polls `GET /bulk-jobs` every **5 s** via `refetchInterval: 5000`. Job status transitions (pending → processing → completed) are reflected within one polling cycle. ### Linting page — fixed polling + immediate invalidation `/linting` polls `GET /lint/results` every **30 s**. After enqueuing a bulk lint operation, the mutation's `onSuccess` immediately invalidates `['lintResults']` and `['videos']` for a faster first update. ### Implication Because completion detection is polling-based, the UI does not reflect job results in real time — there is always a latency of up to one polling interval. For sync jobs that take longer than 8 s, the video editor in particular may lag. If tighter feedback is needed in future, SSE on the sync queue endpoint would be the natural addition. --- ## Job ID Deduplication - Lint-once: `jobId: lint-{videoId}` — BullMQ ignores duplicate job IDs, so submitting the same video twice before the job runs only creates one job - Forced rerun: `jobId: lint-{videoId}-{timestamp}` — always creates a new job ## Related - [[02 - Backend]] - [[05 - Local Setup]] (Redis setup)