fix: add timeouts

Author: zhuravlevma

package-lock.json

@@ -14,7 +14,7 @@
 		"lint": "npx eslint --fix --quiet \"{src,apps,libs,test}/**/*.ts\"",
 		"lint:check": "npx eslint \"{src,apps,libs,test}/**/*.ts\"",
 		"start": "nest start",
-		"start:dev": "npx dotenv nest start --watch",
+		"start:dev": "nest start --watch",
 		"start:debug": "nest start --debug --watch",
 		"start:prod": "node dist/main",
 		"test": "jest",
@@ -60,7 +60,6 @@
 		"@types/supertest": "^6.0.2",
 		"@typescript-eslint/eslint-plugin": "^8.18.2",
 		"@typescript-eslint/parser": "^8.18.2",
-		"dotenv-cli": "^8.0.0",
 		"eslint": "^9.17.0",
 		"eslint-config-prettier": "^9.1.0",
 		"eslint-plugin-prettier": "^5.2.1",

package.json

@@ -10,6 +10,12 @@ import { SagaStep } from './saga-step.model';
 import { DomainMessage } from 'src/__lib__/domain-message';
 import { MessageOrmEntity } from 'src/__relay__/message.orm-entity';
 
+export enum SagaStatusEnum {
+  PENDING = 'PENDING',
+  COMPENSATION = 'COMPENSATION',
+  COMPLETE = 'COMPLETE',
+}
+
 @Entity('sagas')
 export class Saga {
   @PrimaryGeneratedColumn('uuid')
@@ -22,7 +28,7 @@ export class Saga {
   sagaType: string;
 
   @Column()
-  status: string; // InProgress, Completed, Compensated, Failed
+  status: SagaStatusEnum;
 
   @CreateDateColumn()
   startedAt: Date;
@@ -33,6 +39,12 @@ export class Saga {
   @Column({ default: 0 })
   retries: number;
 
+  @CreateDateColumn({ type: 'timestamptz' }) // Добавляем метку времени создания
+  createdAt: Date;
+
+  @UpdateDateColumn({ type: 'timestamptz' }) // Добавляем метку времени обновления
+  updatedAt: Date;
+
   @Column({ nullable: true })
   currentStep: string;
 
@@ -64,6 +76,7 @@ export class Saga {
 
     const newStep = new SagaStep();
     newStep.createCompensationStep(event);
+    this.status = SagaStatusEnum.COMPENSATION;
 
     if (this.steps === undefined) {
       this.steps = [];
@@ -76,6 +89,8 @@ export class Saga {
     const newStep = new SagaStep();
     newStep.createCompleteStep(event);
 
+    this.status = SagaStatusEnum.COMPLETE;
+
     if (this.steps === undefined) {
       this.steps = [];
     }

src/__saga__/models/saga.model.ts

@@ -3,6 +3,7 @@ import { RegistatorService } from './registrator.service';
 import { DomainMessage } from 'src/__lib__/domain-message';
 import { config } from 'src/config';
 import { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
+import { Cron } from '@nestjs/schedule';
 
 export class CreateSagaDto {
   correlationId: string;
@@ -25,4 +26,9 @@ export class RegistatorController {
   async createSaga(@Body() payload: CreateSagaDto) {
     return this.registatorService.createSaga(payload.correlationId);
   }
+
+  @Cron('* * * * *')
+  async timeOut() {
+    await this.registatorService.checkSagaTimeout();
+  }
 }

src/__saga__/registrator.controller.ts

@@ -1,8 +1,10 @@
 import { Injectable, Logger } from '@nestjs/common';
 import { InjectDataSource, InjectRepository } from '@nestjs/typeorm';
-import { DataSource, Repository } from 'typeorm';
-import { Saga } from './models/saga.model';
+import { DataSource, LessThanOrEqual, Repository } from 'typeorm';
+import { Saga, SagaStatusEnum } from './models/saga.model';
 import { DomainMessage } from 'src/__lib__/domain-message';
+import { SagaCompensationEvent } from './saga-compensation.event';
+import { randomUUID } from 'crypto';
 
 @Injectable()
 export class RegistatorService {
@@ -44,6 +46,37 @@ export class RegistatorService {
     return saved;
   }
 
+  async checkSagaTimeout() {
+    const twelveHoursAgo = new Date(Date.now() - 12 * 60 * 60 * 1000);
+    const sagas = await this.sagaRepository.find({
+      where: {
+        status: SagaStatusEnum.PENDING,
+        updatedAt: LessThanOrEqual(twelveHoursAgo),
+      },
+      take: 10,
+    });
+
+    const allMessages = [];
+    for (const saga of sagas) {
+      const messages = saga.compensate(
+        new SagaCompensationEvent({
+          aggregateId: saga.id,
+          saga: {
+            correlationId: randomUUID(),
+            sagaId: saga.id,
+          },
+        }),
+      );
+      for (const msg of messages) {
+        allMessages.push(msg);
+      }
+    }
+    return this.dataSource.transaction(async (transactionalEntityManager) => {
+      await transactionalEntityManager.save(allMessages);
+      return transactionalEntityManager.save(sagas);
+    });
+  }
+
   async createSaga(correlationId: string): Promise<string> {
     this.logger.debug(`Create saga for correlationId ${correlationId}`);
     const saga = await this.sagaRepository.findOne({
@@ -55,7 +88,7 @@ export class RegistatorService {
       const newSaga = new Saga();
       newSaga.correlationId = correlationId;
       newSaga.sagaType = 'compensation';
-      newSaga.status = 'In Progress';
+      newSaga.status = SagaStatusEnum.PENDING;
       newSaga.steps = [];
       const created = await this.sagaRepository.save(newSaga);
       return created.id;