Back to docs hub
Job Module & Workers

Worker operations map cho Threads API v2

Trang này gom toàn bộ producer, queue, worker, QueueEvents và các service đang gọi job. Mục tiêu là mở vào là biết worker nào đang chạy, queue nào đang nhận job, payload đi qua đâu, và những phần nào hiện chỉ có code nhưng chưa được wire đầy đủ.

current inventory
4
BullMQ queues
4
Worker classes
6
Job names
2
Redis list buffers
01

Runtime Stack

Job system dùng BullMQ cho queue/worker chính, Redis client cho một số data structure riêng, và QueueEvents để quan sát trạng thái job.

Producers

src/modules/job

Các class producer expose method enqueue job. Service gọi producer, producer gọi queue.add.

Workers

workers/*.ts

Mỗi worker tạo bằng createWorker, switch theo job.name, concurrency mặc định 3.

QueueEvents

queues/*.ts

Monitor waiting, active, completed, failed cho email/bloom/pinecone. Like chưa có QueueEvents file riêng.

Redis

redisQueueredisWorker

BullMQ dùng IORedis riêng. App cũng dùng redisService cho Bloom filter, blacklist, sets và Redis lists.

02

Worker Matrix

Bảng này là bản đồ chính: queue nào, job name nào, producer nào enqueue, worker nào xử lý, side effect là gì.

BullMQ

EmailWorker

  1. EmailProducer enqueue vào email_queue.
  2. Worker nhận send_verification_email hoặc send_forgot_password_email.
  3. Render EJS template từ thư mục template/.
  4. Gửi mail qua nodemailerService.sendMail.
wired

Được gọi bởi AuthService.forgotPassword và AuthService.resendVerifyEmail.

queue/jobs
email_queue
send_verification_email
send_forgot_password_email
payload
{ userName?, userId?, email, token }
effects
Email verify/reset link tới FRONTEND_URL, log thành công qua logger.
BullMQ + Redis BF

BloomWorker

  1. BloomProducer enqueue generate_bloom sau register.
  2. Worker đảm bảo redisService đã connect.
  3. Gọi BF.ADD filter:usernames và BF.ADD filter:emails.
  4. Validate email/username sau này đọc Bloom bằng BF.EXISTS.
wired

Được gọi bởi AuthService.register. Đây là pipeline availability check nhanh, không phải nguồn chân lý duy nhất.

queue/job
bloom_queue
generate_bloom
payload
{ userName, email }
effects
Ghi Redis Bloom keys filter:usernames, filter:emails.
BullMQ + AI

PineWorker

  1. PostService.create tạo post xong và attach meta trong transaction.
  2. PineProducer.addToPineconeQueue enqueue post_embedding.
  3. Worker gọi mixedBreadService.generateEmbedding.
  4. Upsert vector vào Pinecone namespace posts.
wired

Được gọi sau create post. Search hiện query Pinecone trực tiếp bằng embedding từ query.

queue/job
pinecone_queue
post_embedding
payload
{ content, topic[], postId, userId }
effects
Pinecone record id post:<postId>, metadata gồm postId/userId/content/topics/type.
BullMQ + Redis lists

LikeWorker

  1. Worker support BullMQ jobs sync-post-like và init-sync-job.
  2. PostService.like hiện không gọi LikeProducer; service ghi trực tiếp Redis set và Redis list.
  3. Add like đẩy JSON vào likes_add; unlike đẩy JSON vào likes_remove.
  4. initSyncJob dự kiến batch pop hai list rồi sync MySQL likes/post count.
partially wired

Code tồn tại nhưng entrypoint gọi LikeProducer.initSyncJob chưa thấy trong service/startup hiện tại.

queue/jobs
like_queue
sync-post-like
init-sync-job
redis buffers
likes_add
likes_remove
post:<publicId>:likes
effects
Dự kiến insert likes và tăng/giảm posts.likes_count; hiện có notes ở cuối trang.
03

Service Callers

Đây là nơi job được enqueue từ business flow. Nếu cần tìm “ai gọi worker”, bắt đầu từ bảng này.

AuthService.register
BloomProducer

Gọi addUserNameAndEmailToBloom sau khi tạo user và token pair.

AuthService.forgotPassword
EmailProducer

Gọi sendForgotPasswordEmail song song với lưu verification token hash.

AuthService.resendVerifyEmail
EmailProducer

Gọi sendVerifyEmail song song với lưu verification token hash.

PostService.create
PineProducer

Gọi addToPineconeQueue sau transaction tạo post + topic/mentions.

PostService.like
Redis direct

Không gọi producer; ghi Redis set/count/list trực tiếp để sync sau.

04

Like Pipeline Detail

Like là phần dễ đọc nhầm nhất vì có cả BullMQ producer và Redis list pipeline. Đây là mô tả hiện trạng, không phải đề xuất sửa.

Request path hiện tại

  • Like/unlike gọi PostService.like(publicId, userId, isLiked).
  • Service dùng Redis set post:<publicId>:likes để add/remove userId.
  • Nếu add/remove thành công, service tăng/giảm post:<publicId>:likeCount.
  • Service đẩy JSON vào list likes_add hoặc likes_remove.

Worker path dự kiến

  • LikeWorker.initSyncJob chạy processAddLike và processRemoveLike.
  • Mỗi processor gọi rPopCustomBatch(key, 500), group theo postPublicId.
  • Add path gọi likeRepository.createMany và postRepository.incrementLikedCount.
  • Remove path hiện cũng gọi createMany rồi decrementLikedCount; xem notes bên dưới.
05

Queue Events Monitor

Thư mục queues/ không xử lý job; nó subscribe lifecycle events để log trạng thái queue.

email.queue
email_queue

Log waiting, active, completed, failed.

bloom.queue
bloom_queue

Log waiting, active, completed, failed.

pinecone.queue
pinecone_queue

Log waiting, active, completed, failed.

queues/index
startup import

npm run queue import các QueueEvents này.

like
not present

Chưa có file QueueEvents riêng cho like_queue.

06

Runbook

Cách bật worker/queue monitor hiện tại theo script package.

run workers
npm run worker
# tsx workers/index.ts
# import 4 worker modules
# constructors create Worker instances
run queue events
npm run queue
# tsx queues/index.ts
# subscribe QueueEvents for email/bloom/pinecone
script caveat
npm run all:workers:queues
# currently uses &&
# worker process is long-running
# queue script may not start after it
07

Current Notes & Sharp Edges

Các ghi chú này phản ánh code hiện tại. Theo yêu cầu, tài liệu chỉ ghi lại để team nhìn rõ, không sửa behavior.

like worker

LikeProducer có initSyncJob repeat mỗi 5 giây, nhưng chưa thấy service/startup nào gọi method này. Nếu không enqueue job đó, processAddLike/processRemoveLike sẽ không chạy.

like batch

rPopCustomBatch gọi lMPop nhưng hiện trả về mảng results rỗng, chưa parse dữ liệu từ Redis response.

like remove

processRemoveLike hiện gọi likeRepository.createMany thay vì deleteMany, sau đó decrement count. Đây là behavior đang đọc được từ code.

sync-post-like

syncPostLike đọc rpop post:<publicId>:likes, trong khi request path đang dùng Redis set sAdd/sRem trên cùng key. Phần xử lý chính bên trong method đang comment nhiều.

queue events

QueueEvents dùng constructor mặc định chưa truyền connection từ redisQueue; nó sẽ dựa vào BullMQ default connection behavior.

constants

BLOOM_KEY.FILTER_USERNAME và BLOOM_KEY.FILTER_EMAIL có trailing spaces trong constant, nhưng worker/service hiện hard-code key không có trailing spaces.

email producer

CreateVerifyEmailProducer có userName, nhưng EmailWorker.sendVerificationEmail chỉ dùng email và token.