Navbatlar NestJS'da: @nestjs/bullmq, @Processor va WorkerHost

Qisqacha: @nestjs/bullmq — BullMQ'ning Nest o'rami. BullModule.forRoot({ connection }) Redis ulanishini, BullModule.registerQueue({ name }) navbatni ro'yxatga oladi; producer navbatni @InjectQueue("nom") bilan oladi, ishchi esa @Processor("nom") dekoratorli va WorkerHost dan meros olgan klass — process(job) metodi bilan. Kanon o'lchovida 5 000 ish: xom BullMQ worker 1 126 ms, @nestjs/bullmq 1 120 ms — teng. Yopishda app.close() boshlangan ishni kutadi. vazifalar-api da API faqat ish qo'shadi (Navbatlar provider), ishchi esa 29-qismdagidek alohida jarayon, Nest'siz — unga DI kerak emas va ikki paket kam.

Bu darsda

• BullModule, @InjectQueue, @Processor va WorkerHost bilan Nest navbatini yozasiz.
• @OnWorkerEvent, flow producer va Bull Board'ning Nest'dagi o'rnini bilasiz.
• Nest ishchisi va xom BullMQ worker'ining kanon o'lchovini o'qiysiz.
• vazifalar-api ning producer qismi, outbox va Bull Board himoyasi qanday ulanganini tushuntirasiz.
• Kanon ko'chishda topgan test izolyatsiyasi xatosini ko'rasiz.

Oldin bilishingiz kerak: Nega navbat kerak, BullMQ asoslari, BullMQ ishonchlilik, Transactional outbox naqshi, Dynamic modullar, Monorepo, standalone ilova va CLI buyruqlar.

O'tgan darsda fayllar Nest'ga ko'chdi — @Req() oqimi bilan. Fayl yuklangach, uni qayta ishlash, eslatma yuborish, webhook — bular so'rov ichida emas, fonda bo'lishi kerak. 29-qismda buni BullMQ bilan qilgandik. Endi Nest'da.

1. Nega bu kerak?

29-qismda navbat nima uchun kerakligini ko'rdik (Nega navbat kerak): sekin ish (SMTP, push, tashqi API) foydalanuvchi so'rovini ushlab turmasin. API ishni navbatga qo'yadi va darhol javob beradi. Ishchi (worker) — alohida jarayon — navbatdan olib, bajaradi, xato bo'lsa qayta urinadi.

29-qismdagi ishchi oddiy Node skript edi: new Worker(nom, ishlovchi, sozlama). Ishlovchiga kerak narsalar (pool, pochta, push) qo'lda yaratilib, uzatilgan. Nest'da shu ishni DI bilan qilish mumkin: ishchi klass constructor orqali xizmatlarni oladi, config va lifecycle Nest'niki.

O'xshatish: restoranda ofitsiant buyurtmani oshxona taxtasiga ilib qo'yadi (navbat) va keyingi mijozga o'tadi. Oshpaz taxtadan qog'ozni olib pishiradi (ishchi). Nest'ning o'rami — oshpazga ish joyini tayyorlab beradigan oshxona boshlig'i: pichoq, qozon, mahsulot — hammasi joyida.

2. @nestjs/bullmq asoslari

2.1 Modul va navbat

npm i @nestjs/bullmq bullmq

Kanon tajribasida @nestjs/bullmq 12.0.0 va bullmq 6.3.19 (29-qismdagi bilan bir xil) ishlatildi. Modul:

@Module({
  imports: [
    BullModule.forRoot({
      connection: { host: "127.0.0.1", port: 6379 },
      prefix: "vz",
    }),
    BullModule.registerQueue({ name: "eslatmalar" }),
  ],
  providers: [EslatmaIshchisi, VazifaXizmati],
})
export class NavbatModule {}

• BullModule.forRoot — butun ilova uchun Redis ulanishi va umumiy sozlamalar (prefix, defaultJobOptions). Dinamik modul (Dynamic modullar); config'dan olish uchun — forRootAsync({ inject: [ConfigService], useFactory }).
• BullModule.registerQueue({ name }) — bitta navbat. Har navbat uchun Nest provider yaratadi.

2.2 Producer: @InjectQueue

@Injectable()
export class VazifaXizmati {
  constructor(
    @InjectQueue("eslatmalar") private readonly navbat: Queue,
  ) {}

  async muddatQoy(vazifaId: number, muddat: Date): Promise<void> {
    await this.navbat.add(
      "eslatma",
      { vazifaId },
      {
        delay: muddat.getTime() - Date.now(),
        jobId: `eslatma:${vazifaId}`,
      },
    );
  }
}

Queue — BullMQ'ning o'z klassi; Nest uni yaratib, token orqali beradi (Custom provider'lar). Qolgani 29-qismdagi BullMQ: add(nom, malumot, sozlama), kechiktirilgan ish (delay), takrorni oldini oluvchi jobId (BullMQ asoslari).

2.3 Ishchi: @Processor va WorkerHost

@Processor("eslatmalar", { concurrency: 20 })
export class EslatmaIshchisi extends WorkerHost {
  constructor(private readonly pochta: PochtaXizmati) {
    super();
  }

  async process(job: Job<{ vazifaId: number }>): Promise<void> {
    await this.pochta.eslatmaYubor(job.data.vazifaId);
  }

  @OnWorkerEvent("failed")
  yiqildi(job: Job, xato: Error): void {
    console.error(`ish ${job.id} yiqildi: ${xato.message}`);
  }
}

• @Processor(nom, workerSozlamasi) — klassni navbat ishchisi deb belgilaydi. Ikkinchi argument — BullMQ Worker sozlamalari (concurrency, limiter).
• WorkerHost — asos klass. process(job) ni yozish majburiy. Uning ichida this.worker — haqiqiy BullMQ Worker nusxasi.
• @OnWorkerEvent("failed") — worker hodisasi tinglovchisi: completed, failed, progress, active, stalled.
• constructor — oddiy Nest DI: PochtaXizmati ni boshqa xizmatlar kabi oladi. Bu o'ramning asosiy yutug'i.

@OnWorkerEvent nomlari ishning holatlaridan keladi. 29-qismda ko'rgan hayot yo'lini eslaymiz (BullMQ asoslari):

stateDiagram-v2
  [*] --> waiting: add()
  [*] --> delayed: add() + delay
  delayed --> waiting: vaqti keldi
  waiting --> active: worker oldi
  active --> completed: process() tugadi
  active --> failed: xato, urinish tugadi
  active --> delayed: xato, qayta urinish
  completed --> [*]
  failed --> [*]

O'tishlarda worker hodisa chiqaradi: ish olinganda active, muvaffaqiyatda completed, process() xato tashlaganda failed. Diqqat: failed qayta urinish qolgan bo'lsa ham kelishi mumkin. Ogohlantirishni faqat ish butunlay yiqilganda yuborish uchun hodisa ichida urinishlarni solishtiring: job.attemptsMade >= (job.opts.attempts ?? 1). Muddat qo'yilgan eslatma delayed holatida kutadi — muddatQoy() dagi delay shu.

Ilova ishga tushganda BullModule ning "explorer"i @Processor belgili klasslarni topadi va har biriga Worker yaratadi. app.close() da worker'lar yopiladi (Lifecycle hook'lar).

Lekin app.close() o'zidan chaqirilmaydi. Deploy paytida Docker yoki orkestrator jarayonga SIGTERM signalini yuboradi. Signalni kimdir ushlab, app.close() ni chaqirishi kerak: yoki Nest'ning o'zi (main.ts da app.enableShutdownHooks()), yoki o'z signal ishlovchingiz — vazifalar-api ikkinchisini tanlagan (Lifecycle hook'lar darsida ko'rgandik). Hech biri bo'lmasa, jarayon darhol o'ladi va yarim bajarilgan ish "osilib" qoladi. BullMQ uni keyinroq "stalled" (to'xtab qolgan) deb topib, qayta navbatga qo'yadi (BullMQ ishonchlilik), ya'ni ish ikkinchi marta bajariladi. Eslatma xati ikki marta ketishi — aynan shu.

2.4 Bitta jarayonmi, ikkitami?

@Processor klassini API'ning AppModule iga qo'shsangiz, ishchi API bilan bitta jarayonda ishlaydi. Alohida kirish fayli (ishchi.ts) va standalone kontekst bilan esa — alohida jarayonda. Ikkalasi ham ishlaydi, farqi ishlatishda:

vazifalar-api 29-qismdan beri ikkinchi yo'lda: API faqat ish qo'shadi, ishchi — alohida. Bu darsdagi hamma narsa ikkala holatda ham bir xil yoziladi — farqi faqat @Processor klassi qaysi modulga ulanishida.

2.5 JS modelida

Explorer va yopish g'oyasini kichik modelda ko'ramiz (brauzerda ishlaydi, Redis yo'q):

// @Processor + WorkerHost modeli: "explorer" belgilangan klasslarni
// topib, har biriga worker yaratadi; close() boshlangan ishni kutadi
const PROTSESSORLAR = new Map(); // klass → navbat nomi (metama'lumot)
const Processor = (navbat) => (klass) => {
  PROTSESSORLAR.set(klass, navbat);
  return klass;
};

class EslatmaIshchisi {
  async process(ish) {
    await new Promise((ok) => setTimeout(ok, ish.ms));
    return `eslatma yuborildi: vazifa ${ish.vazifaId}`;
  }
}
Processor("eslatmalar")(EslatmaIshchisi);

function workerYarat(navbat, nusxa, log) {
  const faol = new Set();
  return {
    async qosh(ish) {
      const p = nusxa.process(ish).then((natija) => {
        log.push(`${navbat}: ${natija}`);
      });
      faol.add(p);
      await p;
      faol.delete(p);
    },
    async close() {
      log.push(`${navbat}: yopilmoqda, faol ishlar ${faol.size}`);
      await Promise.all(faol);
      log.push(`${navbat}: yopildi`);
    },
  };
}

// "BullModule explorer": hamma @Processor klasslari uchun worker
const log = [];
const workerlar = [...PROTSESSORLAR].map(([klass, navbat]) =>
  workerYarat(navbat, new klass(), log));

const [w] = workerlar;
const ish = w.qosh({ vazifaId: 9, ms: 50 });
await w.close(); // ish tugaguncha kutadi
await ish;
console.log(log.join("\n"));

Konsolda:

eslatmalar: yopilmoqda, faol ishlar 1
eslatmalar: eslatma yuborildi: vazifa 9
eslatmalar: yopildi

Tartibga qarang: "yopilmoqda" — ish hali tugamagan; "yopildi" — faqat ish tugagandan keyin. Model soddalashtirilgan: dekorator o'rniga funksiya chaqiruvi, Redis o'rniga xotira. Haqiqiy Worker.close() ham shunday: yangi ish olmaydi va faol ishlarni kutadi (BullMQ ishonchlilik).

Tekshirib ko'ring: @Processor("eslatmalar") yozildi, lekin BullModule.registerQueue({ name: "eslatmalar" }) yo'q. @InjectQueue("eslatmalar") ishlaydimi?

3. Flow va Bull Board

3.1 Flow producer

29-qismda flow ni ko'rgandik: ota ish bolalari tugagandan keyin boshlanadi (BullMQ ilg'or imkoniyatlari). Nest'da:

// modulda
BullModule.registerFlowProducer({ name: "hisobot" }),

// xizmatda
constructor(
  @InjectFlowProducer("hisobot") private readonly flow: FlowProducer,
) {}

this.flow.add({ name, queueName, children: [...] }) — 29-qismdagi bilan bir xil API.

3.2 Bull Board

Bull Board — navbatlarni brauzerda ko'rish paneli: kutayotgan, bajarilayotgan, yiqilgan ishlar. Nest uchun @bull-board/nestjs moduli bor (BullBoardModule.forRoot, forFeature). Kanon uni ishlatmadi — 29-qismdagi Express adapteri (@bull-board/express) qoldi. U o'z Express Router'ini beradi va ilova.ts da ulanadi:

// Bull Board — Express adapteri o'z Router'ini beradi (Nest
// controller emas), shuning uchun himoya ham shu yerda: 28 admin roli
const navbat = app.get<NavbatSozlamasi | null>(NAVBATLAR);
if (navbat !== null && a !== null) {
  app.use("/v1/admin/navbat", (req, _res, next) => {
    if (req.sessiya === undefined || req.apiKalit !== undefined) {
      throw new KirilmaganXatosi();
    }
    if (req.sessiya.rol !== "admin") throw new RuxsatYoqXatosi();
    next();
  }, navbatKuzatuv(navbat.navbatlar, "/v1/admin/navbat"));
}

Muhim nuqta: app.use(...) bilan ulangan Express Router'ga Nest guard'lari tegmaydi. Global KirishGuard faqat Nest controller'larini himoya qiladi. Shuning uchun admin tekshiruvi shu yerda, qo'lda. Parcha qisqartirilgan va qatorlarga bo'lingan; tiplar olib tashlangan.

Kanon testi buni tekshiradi (test/nest-integratsiya.test.ts):

test("Bull Board (Express adapteri): oddiy foydalanuvchi 403, " +
  "admin 200 (uchala navbat)", async () => {
  // … aziz — oddiy, bekzod — admin
  await request(http).get("/v1/admin/navbat/api/queues")
    .set("Cookie", aziz).expect(403);
  const r = await request(http).get("/v1/admin/navbat/api/queues")
    .set("Cookie", admin).expect(200);
  expect(r.body.queues.map((q) => q.name).sort())
    .toEqual(["eslatmalar", "tizim", "yetkazishlar"]);
});

Test ikki tomonni tekshiradi. Oddiy foydalanuvchi (Aziz) 403 oladi — demak qo'lda yozilgan admin tekshiruvi ishlayapti. Admin 200 oladi va javobda uchala navbat bor — demak himoya panelning o'zini buzmagan. Bitta test "yopiq" va "ochiq" ni birga qotiradi: kimdir tekshiruvni olib tashlasa ham, kimdir uni haddan tashqari qattiq qilsa ham — test yiqiladi.

Diqqat: Nest ilovasida app.use() bilan Express middleware yoki Router ulasangiz — u Nest himoyasidan tashqarida. Unga alohida himoya kerak va alohida test. Bu xato 28-qism auth'i ko'chganda ham topilgandi: versiyasiz yo'llar hamma controller uchun ochiq qolgan edi.

4. O'lchov: Nest ishchisi va xom worker

4.1 Tajriba

Kanon ikki variantni bitta mashinada, bitta Redis'da solishtirdi:

• xom BullMQ — new Worker(nom, ishlovchi, { concurrency: 20 }) (29-qism kanoni);
• @nestjs/bullmq — @Processor({ name, prefix }, { concurrency: 20 }) + WorkerHost, createApplicationContext ichida (standalone ilova).

Har biriga 5 000 bo'sh ish (addBulk), uch takror, navbat bilan. Ikkalasi ham node-redis klienti bilan — 29-qism qarori. Nest ishchisi (qisqartirilgan parcha):

@Processor({ name: "tC-nest", prefix: "tC" }, { concurrency: CONC })
class NestIshchi extends WorkerHost {
  boshlandi = 0;
  async process(job: Job<{ i: number; kut?: number }>) {
    this.boshlandi++;
    if (job.data.kut !== undefined) {
      await new Promise((ok) => setTimeout(ok, job.data.kut));
    }
    return job.data.i;
  }
}

@Module({
  imports: [
    BullModule.forRoot({ connection: conn, prefix: "tC" }),
    BullModule.registerQueue({ name: "tC-nest" }),
  ],
  providers: [NestIshchi, Qoshuvchi],
})
class IshchiModul {}

conn — 29-qismdagi ulanish(URL): u RedisConnection.clientFactory ni node-redis'ga sozlaydi. Bu BullMQ'ning global sozlamasi — @nestjs/bullmq ham undan foydalandi.

4.2 Natija

Chiqishda uch narsani qidiring: ikki mediana soni, Nest kontekstining ishga tushish vaqti va yopish qatoridagi completed soni.

### 1-2. 5000 ta ish (addBulk), concurrency 20, ishlov — bo'sh; node-redis klienti (clientFactory); 3 takror, navbat bilan
xom BullMQ (Worker)              : 1166, 1126, 1113 ms → mediana 1126
@nestjs/bullmq (@Processor)      : 1120, 1142, 1094 ms → mediana 1120
Nest konteksti ishga tushishi    : 18, 5, 4 ms (createApplicationContext + init: BullModule explorer worker'ni yaratadi)

### 4. Yopish: 800 ms davom etadigan ish boshlangach app.close()
app.close() 792 ms davom etdi; navbat holati: {"completed":1,"active":0,"waiting":0,"failed":0} (completed 1 — ish tugashi kutildi)

Uchta xulosa:

1. Tezlik teng. 1 126 va 1 120 ms — farq o'lchov xatoligi ichida. Nest o'rami har ishga sezilarli narx qo'shmaydi: process() chaqiruvi — oddiy metod chaqiruvi.
2. Ishga tushish arzon. Nest konteksti 4–18 ms da tayyor bo'ldi (bu jarayonda Nest yadrosi allaqachon yuklangan edi — birinchi yuklash narxi bu raqamga kirmagan).
3. Yopish to'g'ri. 800 ms li ish boshlangach app.close() chaqirildi — u 792 ms kutdi va ish completed bo'ldi. Ishchi qayta ishga tushirilganda (deploy) yarim qolgan ish bo'lmaydi.

4.3 Kanon nega xom worker'da qoldi

Tezlik teng bo'lsa, nega Nest variantini olmadik? Kanon qarori (29-qism ishchisi saqlandi):

ishchi.ts ning ishlovchilari (eslatmaIshlovi, yetkazishIshlovi, relay) — bog'liqliklarini argument sifatida oladigan oddiy funksiyalar. Ular uchun DI konteyner foyda bermaydi: hammasi bitta faylda, 20 qatorda yig'iladi. Yangi ishchi ko'paysa yoki ular Nest xizmatlarini (masalan, VazifalarService) qayta ishlatishi kerak bo'lsa — Nest variantiga o'tish oson. Tajriba kodi tayyor.

Qachon Nest ishchisiga o'tish kerak? Uch belgiga qarang. Birinchisi — ishlovchi ilova xizmatini chaqirishi kerak: masalan, eslatmadan oldin VazifalarService orqali vazifa holatini tekshirish. Ikkinchisi — ishlovchilar soni o'sib, ularning bog'liqliklarini qo'lda yig'ish chalkashib ketgan. Uchinchisi — ishchida ham config validatsiyasi, log va yopish tartibi API'dagidek bo'lishi kerak. Uchalasi ham DI'ning kuchli tomoni. vazifalar-api da hozircha birortasi yo'q.

Bu "Nest ishchisi yomon" degani emas. Nest loyihasida ishchi ko'pincha «Ishchi: @Processor va WorkerHost» bo'limidagi shaklda yoziladi — va bu to'g'ri tanlov, agar ishchi ilova xizmatlarini ishlatsa.

5. vazifalar-api da: producer, outbox va ishchi

5.1 API faqat qo'shadi

API'da @nestjs/bullmq yo'q. 29-qismdagi Navbatlar klassi (uchta Queue: eslatmalar, yetkazishlar, tizim) — oddiy provider, token NAVBATLAR. U IntegratsiyaModule da async fabrika bilan yaratiladi (keyingi dars).

Muddat marshruti — MuddatController (qisqartirilgan, qatorlar bo'lingan). Undagi uch narsani belgilab qo'ying: status 202, egasiSifatida (RLS ostidagi tranzaksiya) va navbatga hech qanday murojaat yo'qligi:

@Controller("vazifalar/:id/muddat")
@Cheklov(API_CHEKLOVI)
export class MuddatController {
  constructor(
    @Inject(NAVBATLAR) private readonly o: NavbatSozlamasi,
  ) {}

  @Put()
  @HttpCode(202)
  @JsonTana()
  async qoy(@Egasi() egasi: number, @Req() req: Request) {
    const { id } = tekshir(IdP, req.params, "params");
    const { muddat } = tekshir(MuddatSxemasi, req.body, "body");
    const pool = this.o.pool;
    const yangi = await egasiSifatida(pool, egasi, async (c) => {
      // UPDATE vazifalar SET muddat … RETURNING muddat
      // + o'sha tranzaksiyada: hodisaYoz(c, "vazifa.muddat", …)
    });
    // … topilmasa 404
    const bekor = yangi.muddat === null;
    return { id, muddat: yangi.muddat,
      eslatma: bekor ? "bekor qilinadi" : "rejalanadi" };
  }
}

Diqqat qiling: controller navbatga to'g'ridan yozmaydi. U muddatni va vazifa.muddat hodisasini bitta tranzaksiyada bazaga yozadi — transactional outbox (Transactional outbox naqshi). Ishchi jarayonidagi relay outbox jadvalini o'qib, eslatmani navbatga qo'yadi. Shunday qilib "baza yozildi, navbat yozilmadi" (dual write) muammosi yo'q. 202 Accepted — "qabul qilindi, keyinroq bajariladi".

Tekshirib ko'ring: Redis tez, outbox esa ishchi kelguncha kechikadi. Nega controller navbat.add() ni o'zi chaqirmaydi?

Muddatning butun yo'li ikki jarayon va ikki ombordan o'tadi:

flowchart LR
  A["MuddatController"] --> B["Postgres tranzaksiyasi"]
  B --> C["vazifalar: muddat"]
  B --> D["outbox: vazifa.muddat"]
  D --> E["relay (ishchi jarayoni)"]
  E --> F["Redis: eslatmalar navbati"]
  F --> G["Worker: eslatma xati"]

Chap tomon — API jarayoni, u faqat Postgres'ga yozadi. O'ng tomon — ishchi jarayoni: relay outbox'dan o'qiydi va Redis'ga qo'yadi, worker esa ishni bajaradi. API Redis ishlamay qolsa ham 202 qaytaraveradi: hodisa outbox'da kutadi va Redis qaytgach tarqatiladi. Nest bu zanjirning faqat chap qismiga tegdi — controller. O'ng qism 29-qismdagi kodning o'zi.

5.2 Jonli natija

Kanon tekshiruvidan (haqiqiy Redis va ishchi jarayoni bilan; token yashirilgan, ikki sarlavha qatori olib tashlangan):

$ curl -H Cookie: __Host-sessiya=<token> -X PUT -H Content-Type: application/json -d {"muddat":"2030-01-01T18:00:00+05:00"} http://127.0.0.1:4552/v1/vazifalar/9/muddat
HTTP/1.1 202 Accepted
Content-Type: application/json; charset=utf-8
Content-Length: 67

{"id":9,"muddat":"2030-01-01T13:00:00.000Z","eslatma":"rejalanadi"}

Toshkent vaqti 18:00+05:00 bazada UTC'da saqlandi — 13:00Z. Tekshiruv oxirida outbox jadvalini so'radik. Oxirgi ustunga qarang — u "hali navbatga yetib bormagan hodisalar" soni:

--- outbox (jonli oxirida; ishchi relay)
        tur        | jami | tarqatilmagan 
-------------------+------+---------------
 vazifa.muddat     |    1 |             0
 vazifa.ochirildi  |    1 |             0
 vazifa.yangilandi |    1 |             0
 vazifa.yaratildi  |    2 |             0
(4 rows)

tarqatilmagan 0 — ishchi hamma hodisani olib, tarqatgan. Jadvalda faqat muddat emas, yaratish, yangilash va o'chirish hodisalari ham bor: jonli tekshiruv vazifa ustida to'liq siklni bajardi va har o'zgarish outbox orqali o'tdi. Agar ishchi to'xtab qolsa, aynan shu ustun o'sa boshlaydi — monitoring uchun eng oddiy signal. Ishchi bazaga o'z roli bilan ulanadi (vazifalar_ishchi) — 29-qismdagi eng kam huquq qoidasi.

5.3 Topilma: testlar bitta kanalni bo'lishdi

Integratsiyalar Nest'ga ko'chirilgach, kanonda birinchi to'liq test yugurishida jonli.test boshqa testning hodisasini oldi — outbox.test ning oyna-7 manbali hodisasini. Har test alohida yugurganda — o'tadi, birga — yo'q.

Sabab: ikkala test ham Redis'ning umumiy Pub/Sub kanalini (vz:jonli) tinglaydi. Vitest fayllarni parallel yurgizadi — bir faylning chiqargan hodisasi ikkinchisiga ham yetdi. Kodda xato yo'q: ilovada bitta kanal — to'g'ri dizayn. Xato — test izolyatsiyasida.

Tuzatish — vitest.config.ts da ikki loyiha (projects): bu ikki fayl alohida, ketma-ket (fileParallelism: false), qolganlari parallel:

const KANAL_TESTLARI = ["test/jonli.test.ts", "test/outbox.test.ts"];

export default defineConfig({
  test: {
    // … env, testTimeout, hookTimeout
    projects: [
      { extends: true, test: { name: "asosiy",
        include: ["test/**/*.test.ts", "src/**/*.spec.ts"],
        exclude: KANAL_TESTLARI } },
      { extends: true, test: { name: "redis-kanal",
        include: KANAL_TESTLARI, fileParallelism: false } },
    ],
  },
});

projects — bitta Vitest yugurishida bir nechta "kichik konfiguratsiya": har birining o'z fayllar ro'yxati va sozlamasi bor. fileParallelism: false shu loyiha ichidagi fayllarni birma-bir yurgizadi. Natijada ikki kanal testi hech qachon bir vaqtda ishlamaydi, qolgan yuzlab test esa avvalgidek parallel — umumiy vaqt deyarli o'zgarmaydi.

Muqobil yo'l ham bor edi — har testga o'z kanal prefiksi. Kanon oddiyrog'ini tanladi. Saboq: umumiy tashqi resurs (Redis kanali, navbat nomi, baza jadvali) bilan ishlaydigan testlar parallel yurganda bir-biriga xalal beradi — e2e testlar darsidagi bazani tozalash g'oyasining Redis varianti.

6. Ko'p uchraydigan xatolar

7. Mashqlar

Mashqlar kurs/mashqlar/30/63-bullmq/ da.

1-mashq (oson): qaysi dekorator?

• Redis ulanishi butun ilova uchun — BullModule.
• Navbatni ro'yxatga olish — BullModule.
• Producer navbatni oladi — @("eslatmalar")
• Ishchi klass dekoratori — @("eslatmalar"), asos klass —

2-mashq (o'rta): concurrency modeli

Paketsiz model. yurgiz(ishlar, concurrency) — har ish { id, ms }. Bir vaqtda ko'pi bilan concurrency ta ish bajarilsin (setTimeout bilan). Har ish boshlanganda "boshlandi <id> (faol: N)" ni log'ga yozing. Hammasi tugagach log'ni chiqaring. 5 ta ishni concurrency = 2 bilan sinang. Ishora: navbatdan olish uchun shift(), kutish uchun Promise (Promise asoslari).

async function yurgiz(ishlar, concurrency) {
  const navbat = [...ishlar];
  const log = [];
  let faol = 0;
  async function oqim() {
    while (navbat.length > 0) {
      const ish = navbat.shift();
      faol++;
      log.push(`boshlandi ${ish.id} (faol: ${faol})`);
      await new Promise((ok) => setTimeout(ok, ish.ms));
      faol--;
    }
  }
  await Promise.all(Array.from({ length: concurrency }, oqim));
  return log;
}

const ishlar = [
  { id: 1, ms: 30 }, { id: 2, ms: 10 }, { id: 3, ms: 10 },
  { id: 4, ms: 10 }, { id: 5, ms: 10 },
];
console.log((await yurgiz(ishlar, 2)).join("\n"));

boshlandi 1 (faol: 1)
boshlandi 2 (faol: 2)
boshlandi 3 (faol: 2)
boshlandi 4 (faol: 2)
boshlandi 5 (faol: 2)

3-mashq (qiyin): outbox relay modeli

Paketsiz model. tranzaksiya(baza, fn) — fn baza nusxasida ishlasin; xato bo'lsa o'zgarishlar qo'llanmasin. muddatQoy(baza, id, muddat, xatoBer) — vazifa muddatini va outbox hodisasini yozsin; xatoBer bo'lsa, hodisa yozilgandan keyin xato tashlasin. relay(baza, navbat) — tarqatildi: false hodisalarni navbatga qo'shib, belgilasin. Ikki holatni ko'rsating: muvaffaqiyatli va xatoli muddat qo'yish. Ishora: nusxa uchun structuredClone (Sayoz va chuqur nusxalash).

function tranzaksiya(baza, fn) {
  const nusxa = structuredClone(baza);
  fn(nusxa);
  Object.assign(baza, nusxa); // COMMIT
}

function muddatQoy(baza, id, muddat, xatoBer) {
  tranzaksiya(baza, (t) => {
    t.vazifalar[id].muddat = muddat;
    t.outbox.push({ tur: "vazifa.muddat", id, tarqatildi: false });
    if (xatoBer) throw new Error("baza uzildi");
  });
}

function relay(baza, navbat) {
  for (const h of baza.outbox.filter((x) => !x.tarqatildi)) {
    navbat.push(`eslatma:${h.id}`);
    h.tarqatildi = true;
  }
}

const baza = { vazifalar: { 9: { muddat: null } }, outbox: [] };
const navbat = [];
muddatQoy(baza, 9, "2030-01-01T13:00:00Z", false);
try {
  muddatQoy(baza, 9, "2031-01-01T13:00:00Z", true);
} catch (e) {
  console.log("xato:", e.message);
}
relay(baza, navbat);
console.log("muddat:", baza.vazifalar[9].muddat);
console.log("outbox:", baza.outbox.length, "navbat:", navbat);

xato: baza uzildi
muddat: 2030-01-01T13:00:00Z
outbox: 1 navbat: [ 'eslatma:9' ]

8. Loyiha qadami: producer va ishchi

Bu — 29-qism integratsiyalarini ko'chirishning ikkinchi qismi. Commit — WebSocket gateway'lar darsida.

1. 29-qismdagi navbat modullari o'zgarishsiz ko'chadi: src/navbat/navbat.ts (Navbatlar, ulanish, worker sozlamalari), ishlovchilar.ts, outbox.ts (hodisaYoz, relay), tarqatish.ts, marshrut.ts (navbatKuzatuv — Bull Board).
2. src/ishchi.ts — 29-qismdagi ishchi jarayoni, Nest'siz: config, pool (vazifalar_ishchi roli), yuboruvchilar, worker'lar va relay. SIGTERM — worker.close().
3. Yangi fayl — src/navbat/navbat.controller.ts (MuddatController, «API faqat qo'shadi» bo'limidagi naqsh).
4. src/ilova.ts ga Bull Board — «Bull Board» bo'limidagi parcha (admin tekshiruvi bilan).
5. vitest.config.ts — «Topilma» bo'limidagi projects.
6. package.json ga skript:

"ishchi": "node --env-file-if-exists=.env dist/ishchi.js"

Ishchi ham build'dan ishlaydi — nest build src/ishchi.ts ni ham yig'adi. @nestjs/bullmq o'rnatilmaydi. Tekshiruv — npm run tip, npm run lint. Navbat testlari (navbat, outbox) — Redis bilan, modul ulangandan keyin. Hali commit qilmang.

9. Real ishda

• Nest loyihalarida @nestjs/bullmq — eng keng tarqalgan navbat yechimi: email, rasm qayta ishlash, hisobot. Ko'pincha ishchi xuddi shu ilovada (AppModule ichida) ishlaydi — kichik loyihada qulay, kattasida ishchi alohida jarayonga (standalone kontekst) chiqariladi.
• Bull Board — har BullMQ loyihasida bor. Uni ochiq qoldirish — haqiqiy hodisa: navbatdagi ish ma'lumotida email va telefon bo'ladi.
• Eski @nestjs/bull (Bull 3 uchun) — eski loyihalarda uchraydi; yangi kod — @nestjs/bullmq.
• Intervyu: "Nest'da fon ishi qanday yoziladi?", "Ishchi API bilan bitta jarayonda bo'lishi kerakmi?", "Navbat va baza yozuvini qanday izchil saqlaysiz (outbox)?".

Xulosa

• BullModule.forRoot — ulanish, registerQueue — navbat; producer — @InjectQueue, ishchi — @Processor + WorkerHost.process().
• @OnWorkerEvent, registerFlowProducer + @InjectFlowProducer — 29-qismdagi BullMQ imkoniyatlarining Nest shakli.
• O'lchov: @nestjs/bullmq = xom BullMQ (1 120 / 1 126 ms), node-redis bilan ishlaydi, app.close() faol ishni kutadi.
• Kanon: API'da Navbatlar provider + outbox, ishchi — Nest'siz alohida jarayon (DI kerak emas, +2 paket yo'q).
• app.use bilan ulangan Bull Board Nest guard'idan tashqarida — o'z himoyasi va testi; umumiy Redis kanali testlarni aralashtiradi.

Keyingi dars: Tashqi xizmatlarni Nest modullariga o'rash — S3, SMTP, Redis va Telegram'ni IntegratsiyaModule.forRoot ga yig'amiz: async fabrika, fail-fast va to'g'ri yopish.

Manbalar

• NestJS hujjati — Queues (@nestjs/bullmq, @Processor, WorkerHost, @OnWorkerEvent, flow producer): https://docs.nestjs.com/techniques/queues
• BullMQ hujjati — NestJS: https://docs.bullmq.io/guide/nestjs
• Bull Board — NestJS moduli: https://github.com/felixmosh/bull-board
• Kanon (IlmHamroh, 2026-11-01): @nestjs/bullmq va xom worker tajribasi, MuddatController, Bull Board himoyasi va testi, vitest.config.ts projects, jonli tekshiruv — kanon o'lchovi, bitta mashina.