mirror of
https://github.com/HamzaSha1/zod-backend.git
synced 2026-03-10 17:11:44 +00:00
Merge pull request #92 from Zod-Alkhair/feature/allowance-scheduling-cron-queue
feat: add allowance scheduling with cron, queue, and worker
This commit is contained in:
291
ALLOWANCE_SCHEDULING_SYSTEM.md
Normal file
291
ALLOWANCE_SCHEDULING_SYSTEM.md
Normal file
@ -0,0 +1,291 @@
|
||||
# Allowance Scheduling System (Backend)
|
||||
|
||||
This document captures the complete allowance scheduling feature as implemented in the backend.
|
||||
It is intended as a long-term reference for how the system works, why it was built this way,
|
||||
and how to operate/extend it safely.
|
||||
|
||||
## Goals and Scope
|
||||
|
||||
- Allow parents/guardians to create a recurring allowance schedule for a child.
|
||||
- Credit the allowance automatically based on the schedule.
|
||||
- Scale safely with multiple workers and avoid duplicate credits.
|
||||
- Keep cron lightweight and offload work to RabbitMQ workers.
|
||||
|
||||
Non-goals in the current implementation:
|
||||
- UI changes
|
||||
- Analytics/reporting views
|
||||
- Advanced scheduling (e.g., custom weekdays)
|
||||
|
||||
## High-Level Flow
|
||||
|
||||
1. Parent creates a schedule via API.
|
||||
2. Cron runs every 5 minutes and enqueues due schedules into RabbitMQ.
|
||||
3. Workers consume queue messages, credit the allowance, and update the next run.
|
||||
4. Idempotency is enforced in the database to prevent duplicate credits.
|
||||
|
||||
## Data Model
|
||||
|
||||
### Table: `allowance_schedules`
|
||||
|
||||
Purpose: Stores the schedule definition (amount, frequency, status, next run).
|
||||
|
||||
Key fields:
|
||||
- `guardian_id`, `junior_id`: who funds and who receives.
|
||||
- `amount`: the allowance amount.
|
||||
- `frequency`: DAILY / WEEKLY / MONTHLY.
|
||||
- `status`: ON / OFF.
|
||||
- `next_run_at`, `last_run_at`: scheduling metadata.
|
||||
|
||||
Constraints:
|
||||
- Unique `(guardian_id, junior_id)` ensures one schedule per child.
|
||||
|
||||
Entity: `src/allowance/entities/allowance-schedule.entity.ts`
|
||||
|
||||
### Table: `allowance_credits`
|
||||
|
||||
Purpose: Audit log and idempotency guard for each credit run.
|
||||
|
||||
Key fields:
|
||||
- `schedule_id`: which schedule executed.
|
||||
- `transaction_id`: the resulting transaction (nullable).
|
||||
- `amount`, `run_at`, `credited_at`.
|
||||
|
||||
Idempotency:
|
||||
- Unique `(schedule_id, run_at)` prevents duplicates even with multiple workers.
|
||||
|
||||
Entity: `src/allowance/entities/allowance-credit.entity.ts`
|
||||
|
||||
### Table: `cron_runs`
|
||||
|
||||
Purpose: shared audit table for all cron jobs (not just allowance).
|
||||
|
||||
Key fields:
|
||||
- `job_name`: unique identifier for the cron.
|
||||
- `status`: SUCCESS / FAILED.
|
||||
- `processed_count`: number of schedules processed.
|
||||
- `error_message`: failure reason.
|
||||
- `started_at`, `finished_at`.
|
||||
|
||||
Entity: `src/cron/entities/cron-run.entity.ts`
|
||||
|
||||
## API (Schedule Creation)
|
||||
|
||||
Endpoint:
|
||||
- `POST /guardians/me/allowances/:juniorId`
|
||||
|
||||
Input DTO:
|
||||
- `amount` (required, numeric, positive)
|
||||
- `frequency` (required, enum)
|
||||
- `status` (required, enum)
|
||||
|
||||
DTO: `src/allowance/dtos/request/create-allowance-schedule.request.dto.ts`
|
||||
|
||||
Response DTO:
|
||||
- `AllowanceScheduleResponseDto` with schedule fields.
|
||||
|
||||
DTO: `src/allowance/dtos/response/allowance-schedule.response.dto.ts`
|
||||
|
||||
Business validation:
|
||||
- Child must belong to guardian.
|
||||
- No duplicate schedule for the same guardian+child.
|
||||
|
||||
Service: `src/allowance/services/allowance.service.ts`
|
||||
|
||||
## Cron Producer (Queue Enqueue)
|
||||
|
||||
Cron job:
|
||||
- Runs every 5 minutes.
|
||||
- Batches schedules (cursor-based) to avoid large load.
|
||||
- Enqueues each schedule to RabbitMQ.
|
||||
|
||||
Cron: `src/cron/tasks/allowance-schedule.cron.ts`
|
||||
|
||||
Batch behavior:
|
||||
- Batch size = 100
|
||||
- Uses a cursor (`nextRunAt`, `id`) for stable pagination.
|
||||
- Prevents re-reading the same rows.
|
||||
|
||||
Locking:
|
||||
- `BaseCronService` uses cache lock to ensure only one instance runs.
|
||||
- Each cron run is logged to `cron_runs` with status and processed count.
|
||||
|
||||
Lock: `src/cron/services/base-cron.service.ts`
|
||||
|
||||
## Queue Publisher
|
||||
|
||||
Queue publisher:
|
||||
- Asserts queue + retry/DLQ exchanges.
|
||||
- Enqueues jobs with message id (for traceability).
|
||||
|
||||
Service: `src/allowance/services/allowance-queue.service.ts`
|
||||
|
||||
Queue names:
|
||||
- `allowance.schedule` (main)
|
||||
- `allowance.schedule.retry` (retry with TTL)
|
||||
- `allowance.schedule.dlq` (dead-letter queue)
|
||||
|
||||
Constants: `src/allowance/constants/allowance-queue.constants.ts`
|
||||
|
||||
## Worker Consumer (Transfers)
|
||||
|
||||
Worker:
|
||||
- Consumes `allowance.schedule` queue.
|
||||
- Validates schedule is due and active.
|
||||
- Creates `allowance_credits` record for idempotency.
|
||||
- Transfers money via `cardService.transferToChild`.
|
||||
- Updates `last_run_at` and `next_run_at`.
|
||||
|
||||
Worker: `src/allowance/services/allowance-worker.service.ts`
|
||||
|
||||
Transfer:
|
||||
- Uses existing logic in `card.service.ts` for balance updates and transaction creation.
|
||||
|
||||
Service: `src/card/services/card.service.ts`
|
||||
|
||||
### Idempotency details
|
||||
|
||||
1. Worker inserts `allowance_credits` row first.
|
||||
2. Unique constraint blocks duplicates.
|
||||
3. If transfer fails, the credit row is removed so the job can retry.
|
||||
|
||||
This makes multiple workers safe.
|
||||
|
||||
## Retry + DLQ Strategy
|
||||
|
||||
Retry delay:
|
||||
- Failed jobs are dead-lettered to retry queue.
|
||||
- Retry queue has `messageTtl = 10 minutes`.
|
||||
- After TTL, job is routed back to main queue.
|
||||
|
||||
DLQ:
|
||||
- If a job fails `ALLOWANCE_MAX_RETRIES` times (default 5), it is routed to the DLQ.
|
||||
- This prevents endless loops and allows manual inspection.
|
||||
|
||||
Config:
|
||||
- `ALLOWANCE_MAX_RETRIES` (default 5)
|
||||
|
||||
## Redis Usage
|
||||
|
||||
Redis is used by `BaseCronService` to enforce a **distributed lock** for the cron job.
|
||||
This prevents multiple backend instances from enqueuing the same schedules at the same time.
|
||||
|
||||
Lock behavior:
|
||||
- If the lock key exists, cron exits early.
|
||||
- If the lock key is absent, cron sets it with a TTL and proceeds.
|
||||
- TTL ensures the lock is released even if a node crashes.
|
||||
|
||||
Service: `src/cron/services/base-cron.service.ts`
|
||||
|
||||
## RabbitMQ Setup and Behavior
|
||||
|
||||
The allowance system uses RabbitMQ for asynchronous processing. Cron publishes due schedules,
|
||||
and workers consume them.
|
||||
|
||||
### Exchanges and Queues
|
||||
|
||||
Main queue:
|
||||
- `allowance.schedule`
|
||||
|
||||
Retry setup:
|
||||
- Exchange: `allowance.schedule.retry.exchange`
|
||||
- Queue: `allowance.schedule.retry`
|
||||
- Binding key: `allowance.schedule`
|
||||
- TTL: 10 minutes
|
||||
- Dead-letter route: back to `allowance.schedule`
|
||||
|
||||
Dead-letter queue (DLQ):
|
||||
- Exchange: `allowance.schedule.dlq.exchange`
|
||||
- Queue: `allowance.schedule.dlq`
|
||||
- Binding key: `allowance.schedule`
|
||||
|
||||
### Flow Summary
|
||||
|
||||
1. Cron enqueues jobs to `allowance.schedule`.
|
||||
2. Worker consumes jobs from `allowance.schedule`.
|
||||
3. On failure, job is **dead-lettered** to `allowance.schedule.retry`.
|
||||
4. After 10 minutes, it returns to `allowance.schedule`.
|
||||
5. After max retries, worker publishes the job to `allowance.schedule.dlq`.
|
||||
|
||||
### Where it is configured
|
||||
|
||||
- Publisher setup: `src/allowance/services/allowance-queue.service.ts`
|
||||
- Worker consumer: `src/allowance/services/allowance-worker.service.ts`
|
||||
- Queue constants: `src/allowance/constants/allowance-queue.constants.ts`
|
||||
|
||||
## Environment Variables
|
||||
|
||||
- `RABBITMQ_URL` (required for queue/worker)
|
||||
- `ALLOWANCE_QUEUE_NAME` (optional, defaults to `allowance.schedule`)
|
||||
- `ALLOWANCE_MAX_RETRIES` (optional, defaults to 5)
|
||||
- `ALLOWANCE_RETRY_DELAY_MS` (optional, defaults to 10 minutes)
|
||||
|
||||
### Example .env snippet
|
||||
|
||||
```
|
||||
RABBITMQ_URL=amqp://guest:guest@localhost:5672
|
||||
ALLOWANCE_QUEUE_NAME=allowance.schedule
|
||||
ALLOWANCE_MAX_RETRIES=5
|
||||
ALLOWANCE_RETRY_DELAY_MS=600000
|
||||
```
|
||||
|
||||
## Operational Checklist
|
||||
|
||||
- Ensure Redis is running (cron locking).
|
||||
- Ensure RabbitMQ is running (queue + workers).
|
||||
- Start at least one worker process.
|
||||
- Monitor DLQ for failures.
|
||||
|
||||
## Manual Test Checklist
|
||||
|
||||
1. Create schedule:
|
||||
- POST `/guardians/me/allowances/:juniorId`
|
||||
- Valid amount, frequency, status.
|
||||
2. Duplicate schedule:
|
||||
- Expect `ALLOWANCE.ALREADY_EXISTS`.
|
||||
3. Cron enqueue:
|
||||
- Wait for cron interval or manually trigger.
|
||||
- Confirm messages appear in RabbitMQ.
|
||||
4. Worker:
|
||||
- Ensure worker is running.
|
||||
- Verify transfers happen and `allowance_credits` is created.
|
||||
5. Failure paths:
|
||||
- Simulate transfer failure and verify retry queue behavior.
|
||||
- Confirm DLQ after max retries.
|
||||
|
||||
## Operational Notes
|
||||
|
||||
- For large volumes, scale workers horizontally.
|
||||
- Keep cron lightweight; do not perform transfers in cron.
|
||||
- Monitor queue depth and DLQ entries.
|
||||
|
||||
## Known Limitations (Current)
|
||||
|
||||
- Only one schedule per child (guardian+junior unique).
|
||||
- No custom weekdays or complex schedules.
|
||||
- Retry delay is fixed at 10 minutes (can be configurable).
|
||||
|
||||
## File Map (Quick Reference)
|
||||
|
||||
- API:
|
||||
- `src/allowance/controllers/allowance.controller.ts`
|
||||
- `src/allowance/services/allowance.service.ts`
|
||||
- `src/allowance/dtos/request/create-allowance-schedule.request.dto.ts`
|
||||
- `src/allowance/dtos/response/allowance-schedule.response.dto.ts`
|
||||
|
||||
- Cron:
|
||||
- `src/cron/tasks/allowance-schedule.cron.ts`
|
||||
- `src/cron/services/base-cron.service.ts`
|
||||
|
||||
- Queue/Worker:
|
||||
- `src/allowance/services/allowance-queue.service.ts`
|
||||
- `src/allowance/services/allowance-worker.service.ts`
|
||||
- `src/allowance/constants/allowance-queue.constants.ts`
|
||||
|
||||
- Repositories:
|
||||
- `src/allowance/repositories/allowance-schedule.repository.ts`
|
||||
- `src/allowance/repositories/allowance-credit.repository.ts`
|
||||
|
||||
- Entities:
|
||||
- `src/allowance/entities/allowance-schedule.entity.ts`
|
||||
- `src/allowance/entities/allowance-credit.entity.ts`
|
||||
|
||||
22
src/allowance/allowance.module.ts
Normal file
22
src/allowance/allowance.module.ts
Normal file
@ -0,0 +1,22 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
import { TypeOrmModule } from '@nestjs/typeorm';
|
||||
import { CardModule } from '~/card/card.module';
|
||||
import { JuniorModule } from '~/junior/junior.module';
|
||||
import { AllowanceController } from './controllers';
|
||||
import { AllowanceCredit, AllowanceSchedule } from './entities';
|
||||
import { AllowanceCreditRepository, AllowanceScheduleRepository } from './repositories';
|
||||
import { AllowanceQueueService, AllowanceService, AllowanceWorkerService } from './services';
|
||||
|
||||
@Module({
|
||||
imports: [TypeOrmModule.forFeature([AllowanceSchedule, AllowanceCredit]), JuniorModule, CardModule],
|
||||
controllers: [AllowanceController],
|
||||
providers: [
|
||||
AllowanceService,
|
||||
AllowanceScheduleRepository,
|
||||
AllowanceCreditRepository,
|
||||
AllowanceQueueService,
|
||||
AllowanceWorkerService,
|
||||
],
|
||||
exports: [AllowanceScheduleRepository, AllowanceQueueService, AllowanceCreditRepository],
|
||||
})
|
||||
export class AllowanceModule {}
|
||||
5
src/allowance/constants/allowance-queue.constants.ts
Normal file
5
src/allowance/constants/allowance-queue.constants.ts
Normal file
@ -0,0 +1,5 @@
|
||||
export const ALLOWANCE_QUEUE_NAME = 'allowance.schedule';
|
||||
export const ALLOWANCE_RETRY_QUEUE_NAME = 'allowance.schedule.retry';
|
||||
export const ALLOWANCE_DLQ_NAME = 'allowance.schedule.dlq';
|
||||
export const ALLOWANCE_RETRY_EXCHANGE = 'allowance.schedule.retry.exchange';
|
||||
export const ALLOWANCE_DLQ_EXCHANGE = 'allowance.schedule.dlq.exchange';
|
||||
1
src/allowance/constants/index.ts
Normal file
1
src/allowance/constants/index.ts
Normal file
@ -0,0 +1 @@
|
||||
export * from './allowance-queue.constants';
|
||||
32
src/allowance/controllers/allowance.controller.ts
Normal file
32
src/allowance/controllers/allowance.controller.ts
Normal file
@ -0,0 +1,32 @@
|
||||
import { Body, Controller, Param, Post, UseGuards } from '@nestjs/common';
|
||||
import { ApiBearerAuth, ApiTags } from '@nestjs/swagger';
|
||||
import { Roles } from '~/auth/enums';
|
||||
import { IJwtPayload } from '~/auth/interfaces';
|
||||
import { AllowedRoles, AuthenticatedUser } from '~/common/decorators';
|
||||
import { AccessTokenGuard, RolesGuard } from '~/common/guards';
|
||||
import { ApiDataResponse, ApiLangRequestHeader } from '~/core/decorators';
|
||||
import { ResponseFactory } from '~/core/utils';
|
||||
import { CreateAllowanceScheduleRequestDto } from '../dtos/request';
|
||||
import { AllowanceScheduleResponseDto } from '../dtos/response';
|
||||
import { AllowanceService } from '../services';
|
||||
|
||||
@Controller('guardians/me/allowances')
|
||||
@ApiTags('Allowances')
|
||||
@ApiBearerAuth()
|
||||
@ApiLangRequestHeader()
|
||||
@UseGuards(AccessTokenGuard, RolesGuard)
|
||||
@AllowedRoles(Roles.GUARDIAN)
|
||||
export class AllowanceController {
|
||||
constructor(private readonly allowanceService: AllowanceService) {}
|
||||
|
||||
@Post(':juniorId')
|
||||
@ApiDataResponse(AllowanceScheduleResponseDto)
|
||||
async createSchedule(
|
||||
@AuthenticatedUser() { sub }: IJwtPayload,
|
||||
@Param('juniorId') juniorId: string,
|
||||
@Body() body: CreateAllowanceScheduleRequestDto,
|
||||
) {
|
||||
const schedule = await this.allowanceService.createSchedule(sub, juniorId, body);
|
||||
return ResponseFactory.data(new AllowanceScheduleResponseDto(schedule));
|
||||
}
|
||||
}
|
||||
1
src/allowance/controllers/index.ts
Normal file
1
src/allowance/controllers/index.ts
Normal file
@ -0,0 +1 @@
|
||||
export * from './allowance.controller';
|
||||
@ -0,0 +1,38 @@
|
||||
import { ApiProperty } from '@nestjs/swagger';
|
||||
import { IsEnum, IsNotEmpty, IsNumber, IsPositive } from 'class-validator';
|
||||
import { i18nValidationMessage as i18n } from 'nestjs-i18n';
|
||||
import { AllowanceFrequency, AllowanceScheduleStatus } from '~/allowance/enums';
|
||||
|
||||
export class CreateAllowanceScheduleRequestDto {
|
||||
@ApiProperty({ example: 400 })
|
||||
@IsNotEmpty({
|
||||
message: i18n('validation.IsNotEmpty', { path: 'general', property: 'allowance.amount' }),
|
||||
})
|
||||
@IsNumber(
|
||||
{ maxDecimalPlaces: 2 },
|
||||
{ message: i18n('validation.IsNumber', { path: 'general', property: 'allowance.amount' }) },
|
||||
)
|
||||
@IsPositive({
|
||||
message: i18n('validation.IsPositive', { path: 'general', property: 'allowance.amount' }),
|
||||
})
|
||||
amount!: number;
|
||||
|
||||
@ApiProperty({ enum: AllowanceFrequency, example: AllowanceFrequency.WEEKLY })
|
||||
@IsNotEmpty({
|
||||
message: i18n('validation.IsNotEmpty', { path: 'general', property: 'allowance.frequency' }),
|
||||
})
|
||||
@IsEnum(AllowanceFrequency, {
|
||||
message: i18n('validation.IsEnum', { path: 'general', property: 'allowance.frequency' }),
|
||||
})
|
||||
frequency!: AllowanceFrequency;
|
||||
|
||||
@ApiProperty({ enum: AllowanceScheduleStatus, example: AllowanceScheduleStatus.ON })
|
||||
@IsNotEmpty({
|
||||
message: i18n('validation.IsNotEmpty', { path: 'general', property: 'allowance.status' }),
|
||||
})
|
||||
@IsEnum(AllowanceScheduleStatus, {
|
||||
message: i18n('validation.IsEnum', { path: 'general', property: 'allowance.status' }),
|
||||
})
|
||||
status!: AllowanceScheduleStatus;
|
||||
|
||||
}
|
||||
1
src/allowance/dtos/request/index.ts
Normal file
1
src/allowance/dtos/request/index.ts
Normal file
@ -0,0 +1 @@
|
||||
export * from './create-allowance-schedule.request.dto';
|
||||
@ -0,0 +1,48 @@
|
||||
import { ApiProperty } from '@nestjs/swagger';
|
||||
import { AllowanceSchedule } from '~/allowance/entities/allowance-schedule.entity';
|
||||
import { AllowanceFrequency, AllowanceScheduleStatus } from '~/allowance/enums';
|
||||
|
||||
export class AllowanceScheduleResponseDto {
|
||||
@ApiProperty({ example: '123e4567-e89b-12d3-a456-426614174000' })
|
||||
id!: string;
|
||||
|
||||
@ApiProperty({ example: '123e4567-e89b-12d3-a456-426614174000' })
|
||||
guardianId!: string;
|
||||
|
||||
@ApiProperty({ example: '123e4567-e89b-12d3-a456-426614174000' })
|
||||
juniorId!: string;
|
||||
|
||||
@ApiProperty({ example: 400 })
|
||||
amount!: number;
|
||||
|
||||
@ApiProperty({ enum: AllowanceFrequency, example: AllowanceFrequency.WEEKLY })
|
||||
frequency!: AllowanceFrequency;
|
||||
|
||||
@ApiProperty({ enum: AllowanceScheduleStatus, example: AllowanceScheduleStatus.ON })
|
||||
status!: AllowanceScheduleStatus;
|
||||
|
||||
@ApiProperty({ example: '2026-01-01T00:00:00.000Z' })
|
||||
nextRunAt!: Date;
|
||||
|
||||
@ApiProperty({ example: null })
|
||||
lastRunAt!: Date | null;
|
||||
|
||||
@ApiProperty({ example: '2026-01-01T00:00:00.000Z' })
|
||||
createdAt!: Date;
|
||||
|
||||
@ApiProperty({ example: '2026-01-01T00:00:00.000Z' })
|
||||
updatedAt!: Date;
|
||||
|
||||
constructor(data: AllowanceSchedule) {
|
||||
this.id = data.id;
|
||||
this.guardianId = data.guardianId;
|
||||
this.juniorId = data.juniorId;
|
||||
this.amount = Number(data.amount);
|
||||
this.frequency = data.frequency;
|
||||
this.status = data.status;
|
||||
this.nextRunAt = data.nextRunAt;
|
||||
this.lastRunAt = data.lastRunAt;
|
||||
this.createdAt = data.createdAt;
|
||||
this.updatedAt = data.updatedAt;
|
||||
}
|
||||
}
|
||||
1
src/allowance/dtos/response/index.ts
Normal file
1
src/allowance/dtos/response/index.ts
Normal file
@ -0,0 +1 @@
|
||||
export * from './allowance-schedule.response.dto';
|
||||
42
src/allowance/entities/allowance-credit.entity.ts
Normal file
42
src/allowance/entities/allowance-credit.entity.ts
Normal file
@ -0,0 +1,42 @@
|
||||
import {
|
||||
BaseEntity,
|
||||
Column,
|
||||
CreateDateColumn,
|
||||
Entity,
|
||||
JoinColumn,
|
||||
ManyToOne,
|
||||
PrimaryGeneratedColumn,
|
||||
} from 'typeorm';
|
||||
import { Transaction } from '~/card/entities/transaction.entity';
|
||||
import { AllowanceSchedule } from './allowance-schedule.entity';
|
||||
|
||||
@Entity('allowance_credits')
|
||||
export class AllowanceCredit extends BaseEntity {
|
||||
@PrimaryGeneratedColumn('uuid')
|
||||
id!: string;
|
||||
|
||||
@Column({ type: 'decimal', precision: 12, scale: 2, name: 'amount' })
|
||||
amount!: number;
|
||||
|
||||
@Column({ type: 'timestamp with time zone', name: 'run_at' })
|
||||
runAt!: Date;
|
||||
|
||||
@CreateDateColumn({ type: 'timestamp with time zone', default: () => 'CURRENT_TIMESTAMP', name: 'credited_at' })
|
||||
creditedAt!: Date;
|
||||
|
||||
@Column({ type: 'uuid', name: 'schedule_id' })
|
||||
scheduleId!: string;
|
||||
|
||||
|
||||
@ManyToOne(() => AllowanceSchedule, (schedule) => schedule.credits, { onDelete: 'CASCADE' })
|
||||
@JoinColumn({ name: 'schedule_id' })
|
||||
schedule!: AllowanceSchedule;
|
||||
|
||||
@Column({ type: 'uuid', name: 'transaction_id', nullable: true })
|
||||
transactionId!: string | null;
|
||||
|
||||
@ManyToOne(() => Transaction, { onDelete: 'SET NULL', nullable: true })
|
||||
@JoinColumn({ name: 'transaction_id' })
|
||||
transaction!: Transaction | null;
|
||||
|
||||
}
|
||||
60
src/allowance/entities/allowance-schedule.entity.ts
Normal file
60
src/allowance/entities/allowance-schedule.entity.ts
Normal file
@ -0,0 +1,60 @@
|
||||
import {
|
||||
BaseEntity,
|
||||
Column,
|
||||
CreateDateColumn,
|
||||
Entity,
|
||||
JoinColumn,
|
||||
ManyToOne,
|
||||
OneToMany,
|
||||
PrimaryGeneratedColumn,
|
||||
UpdateDateColumn,
|
||||
} from 'typeorm';
|
||||
import { Guardian } from '~/guardian/entities/guradian.entity';
|
||||
import { Junior } from '~/junior/entities';
|
||||
import { AllowanceFrequency, AllowanceScheduleStatus } from '../enums';
|
||||
import { AllowanceCredit } from './allowance-credit.entity';
|
||||
|
||||
@Entity('allowance_schedules')
|
||||
export class AllowanceSchedule extends BaseEntity {
|
||||
@PrimaryGeneratedColumn('uuid')
|
||||
id!: string;
|
||||
|
||||
@Column({ type: 'decimal', precision: 12, scale: 2, name: 'amount' })
|
||||
amount!: number;
|
||||
|
||||
@Column({ type: 'varchar', name: 'frequency' })
|
||||
frequency!: AllowanceFrequency;
|
||||
|
||||
@Column({ type: 'varchar', name: 'status', default: AllowanceScheduleStatus.ON })
|
||||
status!: AllowanceScheduleStatus;
|
||||
|
||||
@Column({ type: 'timestamp with time zone', name: 'next_run_at' })
|
||||
nextRunAt!: Date;
|
||||
|
||||
@Column({ type: 'timestamp with time zone', name: 'last_run_at', nullable: true })
|
||||
lastRunAt!: Date | null;
|
||||
|
||||
@CreateDateColumn({ type: 'timestamp with time zone', default: () => 'CURRENT_TIMESTAMP', name: 'created_at' })
|
||||
createdAt!: Date;
|
||||
|
||||
@UpdateDateColumn({ type: 'timestamp with time zone', default: () => 'CURRENT_TIMESTAMP', name: 'updated_at' })
|
||||
updatedAt!: Date;
|
||||
|
||||
@Column({ type: 'uuid', name: 'guardian_id' })
|
||||
guardianId!: string;
|
||||
|
||||
@ManyToOne(() => Guardian, (guardian) => guardian.allowanceSchedules, { onDelete: 'CASCADE' })
|
||||
@JoinColumn({ name: 'guardian_id' })
|
||||
guardian!: Guardian;
|
||||
|
||||
@Column({ type: 'uuid', name: 'junior_id' })
|
||||
juniorId!: string;
|
||||
|
||||
@ManyToOne(() => Junior, (junior) => junior.allowanceSchedules, { onDelete: 'CASCADE' })
|
||||
@JoinColumn({ name: 'junior_id' })
|
||||
junior!: Junior;
|
||||
|
||||
@OneToMany(() => AllowanceCredit, (credit) => credit.schedule)
|
||||
credits!: AllowanceCredit[];
|
||||
|
||||
}
|
||||
2
src/allowance/entities/index.ts
Normal file
2
src/allowance/entities/index.ts
Normal file
@ -0,0 +1,2 @@
|
||||
export * from './allowance-credit.entity';
|
||||
export * from './allowance-schedule.entity';
|
||||
5
src/allowance/enums/allowance-frequency.enum.ts
Normal file
5
src/allowance/enums/allowance-frequency.enum.ts
Normal file
@ -0,0 +1,5 @@
|
||||
export enum AllowanceFrequency {
|
||||
DAILY = 'DAILY',
|
||||
WEEKLY = 'WEEKLY',
|
||||
MONTHLY = 'MONTHLY',
|
||||
}
|
||||
4
src/allowance/enums/allowance-schedule-status.enum.ts
Normal file
4
src/allowance/enums/allowance-schedule-status.enum.ts
Normal file
@ -0,0 +1,4 @@
|
||||
export enum AllowanceScheduleStatus {
|
||||
ON = 'ON',
|
||||
OFF = 'OFF',
|
||||
}
|
||||
2
src/allowance/enums/index.ts
Normal file
2
src/allowance/enums/index.ts
Normal file
@ -0,0 +1,2 @@
|
||||
export * from './allowance-frequency.enum';
|
||||
export * from './allowance-schedule-status.enum';
|
||||
32
src/allowance/repositories/allowance-credit.repository.ts
Normal file
32
src/allowance/repositories/allowance-credit.repository.ts
Normal file
@ -0,0 +1,32 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { Repository } from 'typeorm';
|
||||
import { AllowanceCredit } from '~/allowance/entities';
|
||||
|
||||
@Injectable()
|
||||
export class AllowanceCreditRepository {
|
||||
constructor(
|
||||
@InjectRepository(AllowanceCredit)
|
||||
private readonly allowanceCreditRepository: Repository<AllowanceCredit>,
|
||||
) {}
|
||||
|
||||
createCredit(scheduleId: string, amount: number, runAt: Date): Promise<AllowanceCredit> {
|
||||
return this.allowanceCreditRepository.save(
|
||||
this.allowanceCreditRepository.create({
|
||||
scheduleId,
|
||||
amount,
|
||||
runAt,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
findByScheduleAndRunAt(scheduleId: string, runAt: Date): Promise<AllowanceCredit | null> {
|
||||
return this.allowanceCreditRepository.findOne({
|
||||
where: { scheduleId, runAt },
|
||||
});
|
||||
}
|
||||
|
||||
deleteById(id: string): Promise<void> {
|
||||
return this.allowanceCreditRepository.delete({ id }).then(() => undefined);
|
||||
}
|
||||
}
|
||||
67
src/allowance/repositories/allowance-schedule.repository.ts
Normal file
67
src/allowance/repositories/allowance-schedule.repository.ts
Normal file
@ -0,0 +1,67 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { Repository } from 'typeorm';
|
||||
import { AllowanceSchedule } from '~/allowance/entities/allowance-schedule.entity';
|
||||
import { CreateAllowanceScheduleRequestDto } from '~/allowance/dtos/request';
|
||||
import { AllowanceScheduleStatus } from '~/allowance/enums';
|
||||
|
||||
@Injectable()
|
||||
export class AllowanceScheduleRepository {
|
||||
constructor(
|
||||
@InjectRepository(AllowanceSchedule)
|
||||
private readonly allowanceScheduleRepository: Repository<AllowanceSchedule>,
|
||||
) {}
|
||||
|
||||
findByGuardianAndJunior(guardianId: string, juniorId: string): Promise<AllowanceSchedule | null> {
|
||||
return this.allowanceScheduleRepository.findOne({
|
||||
where: { guardianId, juniorId },
|
||||
});
|
||||
}
|
||||
|
||||
createSchedule(guardianId: string, juniorId: string, body: CreateAllowanceScheduleRequestDto, nextRunAt: Date) {
|
||||
return this.allowanceScheduleRepository.save(
|
||||
this.allowanceScheduleRepository.create({
|
||||
guardianId,
|
||||
juniorId,
|
||||
amount: body.amount,
|
||||
frequency: body.frequency,
|
||||
status: body.status,
|
||||
nextRunAt,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
findDueSchedulesBatch(
|
||||
limit: number,
|
||||
cursor?: { nextRunAt: Date; id: string },
|
||||
): Promise<AllowanceSchedule[]> {
|
||||
const query = this.allowanceScheduleRepository
|
||||
.createQueryBuilder('schedule')
|
||||
.where('schedule.status = :status', { status: AllowanceScheduleStatus.ON })
|
||||
.andWhere('schedule.nextRunAt <= :now', { now: new Date() });
|
||||
|
||||
if (cursor) {
|
||||
query.andWhere(
|
||||
'(schedule.nextRunAt > :cursorDate OR (schedule.nextRunAt = :cursorDate AND schedule.id > :cursorId))',
|
||||
{
|
||||
cursorDate: cursor.nextRunAt,
|
||||
cursorId: cursor.id,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
return query
|
||||
.orderBy('schedule.nextRunAt', 'ASC')
|
||||
.addOrderBy('schedule.id', 'ASC')
|
||||
.take(limit)
|
||||
.getMany();
|
||||
}
|
||||
|
||||
findById(id: string): Promise<AllowanceSchedule | null> {
|
||||
return this.allowanceScheduleRepository.findOne({ where: { id } });
|
||||
}
|
||||
|
||||
updateScheduleRun(id: string, lastRunAt: Date, nextRunAt: Date) {
|
||||
return this.allowanceScheduleRepository.update({ id }, { lastRunAt, nextRunAt });
|
||||
}
|
||||
}
|
||||
2
src/allowance/repositories/index.ts
Normal file
2
src/allowance/repositories/index.ts
Normal file
@ -0,0 +1,2 @@
|
||||
export * from './allowance-credit.repository';
|
||||
export * from './allowance-schedule.repository';
|
||||
88
src/allowance/services/allowance-queue.service.ts
Normal file
88
src/allowance/services/allowance-queue.service.ts
Normal file
@ -0,0 +1,88 @@
|
||||
import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common';
|
||||
import { ConfigService } from '@nestjs/config';
|
||||
import { AmqpConnectionManager, ChannelWrapper, connect } from 'amqp-connection-manager';
|
||||
import {
|
||||
ALLOWANCE_DLQ_EXCHANGE,
|
||||
ALLOWANCE_DLQ_NAME,
|
||||
ALLOWANCE_QUEUE_NAME,
|
||||
ALLOWANCE_RETRY_EXCHANGE,
|
||||
ALLOWANCE_RETRY_QUEUE_NAME,
|
||||
} from '../constants';
|
||||
|
||||
@Injectable()
|
||||
export class AllowanceQueueService implements OnModuleDestroy {
|
||||
private readonly logger = new Logger(AllowanceQueueService.name);
|
||||
private connection?: AmqpConnectionManager;
|
||||
private channel?: ChannelWrapper;
|
||||
private readonly queueName: string;
|
||||
private readonly rabbitUrl?: string;
|
||||
private readonly retryDelayMs: number;
|
||||
|
||||
constructor(private readonly configService: ConfigService) {
|
||||
this.queueName = this.configService.get<string>('ALLOWANCE_QUEUE_NAME') || ALLOWANCE_QUEUE_NAME;
|
||||
this.rabbitUrl = this.configService.get<string>('RABBITMQ_URL');
|
||||
this.retryDelayMs = Number(this.configService.get<string>('ALLOWANCE_RETRY_DELAY_MS') || 10 * 60 * 1000);
|
||||
}
|
||||
|
||||
async enqueueSchedule(scheduleId: string, runAt: Date): Promise<void> {
|
||||
if (!this.rabbitUrl) {
|
||||
this.logger.warn('RABBITMQ_URL is not set; skipping allowance enqueue.');
|
||||
return;
|
||||
}
|
||||
|
||||
if (!this.connection || !this.channel) {
|
||||
this.connection = connect([this.rabbitUrl]);
|
||||
this.connection.on('connect', () => {
|
||||
this.logger.log('RabbitMQ connected (publisher).');
|
||||
});
|
||||
this.connection.on('disconnect', (params) => {
|
||||
this.logger.error(`RabbitMQ disconnected (publisher): ${params?.err?.message || 'unknown error'}`);
|
||||
});
|
||||
this.channel = this.connection.createChannel({
|
||||
setup: async (channel: any) => {
|
||||
await channel.assertExchange(ALLOWANCE_RETRY_EXCHANGE, 'direct', { durable: true });
|
||||
await channel.assertExchange(ALLOWANCE_DLQ_EXCHANGE, 'direct', { durable: true });
|
||||
|
||||
await channel.assertQueue(ALLOWANCE_RETRY_QUEUE_NAME, {
|
||||
durable: true,
|
||||
messageTtl: this.retryDelayMs,
|
||||
deadLetterExchange: '',
|
||||
deadLetterRoutingKey: this.queueName,
|
||||
});
|
||||
|
||||
await channel.assertQueue(ALLOWANCE_DLQ_NAME, {
|
||||
durable: true,
|
||||
});
|
||||
|
||||
await channel.bindQueue(ALLOWANCE_RETRY_QUEUE_NAME, ALLOWANCE_RETRY_EXCHANGE, this.queueName);
|
||||
await channel.bindQueue(ALLOWANCE_DLQ_NAME, ALLOWANCE_DLQ_EXCHANGE, this.queueName);
|
||||
|
||||
await channel.assertQueue(this.queueName, {
|
||||
durable: true,
|
||||
deadLetterExchange: ALLOWANCE_RETRY_EXCHANGE,
|
||||
deadLetterRoutingKey: this.queueName,
|
||||
});
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
const payload = {
|
||||
scheduleId,
|
||||
runAt: runAt.toISOString(),
|
||||
};
|
||||
|
||||
const messageId = `allowance:${scheduleId}:${runAt.toISOString()}`;
|
||||
const options: any = {
|
||||
persistent: true,
|
||||
messageId,
|
||||
contentType: 'application/json',
|
||||
};
|
||||
|
||||
await this.channel.sendToQueue(this.queueName, Buffer.from(JSON.stringify(payload)), options);
|
||||
}
|
||||
|
||||
async onModuleDestroy() {
|
||||
await this.channel?.close();
|
||||
await this.connection?.close();
|
||||
}
|
||||
}
|
||||
162
src/allowance/services/allowance-worker.service.ts
Normal file
162
src/allowance/services/allowance-worker.service.ts
Normal file
@ -0,0 +1,162 @@
|
||||
import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
|
||||
import { ConfigService } from '@nestjs/config';
|
||||
import { AmqpConnectionManager, ChannelWrapper, connect } from 'amqp-connection-manager';
|
||||
import moment from 'moment';
|
||||
import { CardService } from '~/card/services';
|
||||
import { AllowanceScheduleRepository } from '../repositories/allowance-schedule.repository';
|
||||
import { AllowanceCreditRepository } from '../repositories/allowance-credit.repository';
|
||||
import { ALLOWANCE_DLQ_EXCHANGE, ALLOWANCE_QUEUE_NAME } from '../constants';
|
||||
import { AllowanceFrequency, AllowanceScheduleStatus } from '../enums';
|
||||
|
||||
type AllowanceQueuePayload = {
|
||||
scheduleId: string;
|
||||
runAt: string;
|
||||
};
|
||||
|
||||
@Injectable()
|
||||
export class AllowanceWorkerService implements OnModuleInit, OnModuleDestroy {
|
||||
private readonly logger = new Logger(AllowanceWorkerService.name);
|
||||
private connection?: AmqpConnectionManager;
|
||||
private channel?: ChannelWrapper;
|
||||
private readonly queueName: string;
|
||||
private readonly rabbitUrl?: string;
|
||||
private readonly maxRetries: number;
|
||||
|
||||
constructor(
|
||||
private readonly configService: ConfigService,
|
||||
private readonly allowanceScheduleRepository: AllowanceScheduleRepository,
|
||||
private readonly allowanceCreditRepository: AllowanceCreditRepository,
|
||||
private readonly cardService: CardService,
|
||||
) {
|
||||
this.queueName = this.configService.get<string>('ALLOWANCE_QUEUE_NAME') || ALLOWANCE_QUEUE_NAME;
|
||||
this.rabbitUrl = this.configService.get<string>('RABBITMQ_URL');
|
||||
this.maxRetries = Number(this.configService.get<string>('ALLOWANCE_MAX_RETRIES') || 5);
|
||||
}
|
||||
|
||||
async onModuleInit() {
|
||||
if (!this.rabbitUrl) {
|
||||
this.logger.warn('RABBITMQ_URL is not set; allowance worker is disabled.');
|
||||
return;
|
||||
}
|
||||
|
||||
this.connection = connect([this.rabbitUrl]);
|
||||
this.connection.on('connect', () => {
|
||||
this.logger.log('RabbitMQ connected (worker).');
|
||||
});
|
||||
this.connection.on('disconnect', (params) => {
|
||||
this.logger.error(`RabbitMQ disconnected (worker): ${params?.err?.message || 'unknown error'}`);
|
||||
});
|
||||
this.channel = this.connection.createChannel({
|
||||
setup: async (channel: any) => {
|
||||
await channel.assertQueue(this.queueName, { durable: true });
|
||||
await channel.prefetch(10);
|
||||
await channel.consume(this.queueName, (msg: any) => this.handleMessage(channel, msg), {
|
||||
noAck: false,
|
||||
});
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async onModuleDestroy() {
|
||||
await this.channel?.close();
|
||||
await this.connection?.close();
|
||||
}
|
||||
|
||||
private async handleMessage(channel: any, msg: any) {
|
||||
if (!msg) {
|
||||
return;
|
||||
}
|
||||
|
||||
let payload: AllowanceQueuePayload;
|
||||
try {
|
||||
payload = JSON.parse(msg.content.toString()) as AllowanceQueuePayload;
|
||||
} catch (error) {
|
||||
const stack = error instanceof Error ? error.stack : undefined;
|
||||
this.logger.error('Invalid allowance queue payload', stack || error);
|
||||
channel.ack(msg);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await this.processAllowanceJob(payload);
|
||||
channel.ack(msg);
|
||||
} catch (error) {
|
||||
const stack = error instanceof Error ? error.stack : undefined;
|
||||
this.logger.error(`Allowance job failed for schedule ${payload.scheduleId}`, stack || error);
|
||||
const retryCount = this.getRetryCount(msg);
|
||||
if (retryCount >= this.maxRetries) {
|
||||
this.logger.error(`Allowance job exceeded max retries (${this.maxRetries}). Sending to DLQ.`);
|
||||
channel.publish(ALLOWANCE_DLQ_EXCHANGE, this.queueName, msg.content, {
|
||||
contentType: msg.properties.contentType,
|
||||
messageId: msg.properties.messageId,
|
||||
headers: msg.properties.headers,
|
||||
persistent: true,
|
||||
});
|
||||
channel.ack(msg);
|
||||
return;
|
||||
}
|
||||
channel.nack(msg, false, false);
|
||||
}
|
||||
}
|
||||
|
||||
private async processAllowanceJob(payload: AllowanceQueuePayload): Promise<void> {
|
||||
const runAt = new Date(payload.runAt);
|
||||
if (!payload.scheduleId || Number.isNaN(runAt.getTime())) {
|
||||
return;
|
||||
}
|
||||
|
||||
const schedule = await this.allowanceScheduleRepository.findById(payload.scheduleId);
|
||||
if (!schedule) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (schedule.status !== AllowanceScheduleStatus.ON) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (schedule.nextRunAt > runAt) {
|
||||
return;
|
||||
}
|
||||
|
||||
let credit = null;
|
||||
try {
|
||||
credit = await this.allowanceCreditRepository.createCredit(schedule.id, schedule.amount, runAt);
|
||||
} catch (error: any) {
|
||||
if (error?.code === '23505') {
|
||||
return;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
||||
try {
|
||||
await this.cardService.transferToChild(schedule.juniorId, schedule.amount);
|
||||
const nextRunAt = this.computeNextRunAt(schedule.frequency);
|
||||
await this.allowanceScheduleRepository.updateScheduleRun(schedule.id, runAt, nextRunAt);
|
||||
} catch (error) {
|
||||
if (credit) {
|
||||
await this.allowanceCreditRepository.deleteById(credit.id);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private computeNextRunAt(frequency: AllowanceFrequency): Date {
|
||||
const base = moment();
|
||||
switch (frequency) {
|
||||
case AllowanceFrequency.DAILY:
|
||||
return base.add(1, 'day').toDate();
|
||||
case AllowanceFrequency.WEEKLY:
|
||||
return base.add(1, 'week').toDate();
|
||||
case AllowanceFrequency.MONTHLY:
|
||||
return base.add(1, 'month').toDate();
|
||||
default:
|
||||
return base.toDate();
|
||||
}
|
||||
}
|
||||
|
||||
private getRetryCount(msg: any): number {
|
||||
const deaths = (msg.properties.headers?.['x-death'] as Array<{ count?: number }> | undefined) || [];
|
||||
const retryDeath = deaths.find((death) => death?.count != null);
|
||||
return retryDeath?.count ?? 0;
|
||||
}
|
||||
}
|
||||
55
src/allowance/services/allowance.service.ts
Normal file
55
src/allowance/services/allowance.service.ts
Normal file
@ -0,0 +1,55 @@
|
||||
import { BadRequestException, Injectable, Logger } from '@nestjs/common';
|
||||
import moment from 'moment';
|
||||
import { JuniorService } from '~/junior/services';
|
||||
import { CreateAllowanceScheduleRequestDto } from '../dtos/request';
|
||||
import { AllowanceSchedule } from '../entities/allowance-schedule.entity';
|
||||
import { AllowanceFrequency, AllowanceScheduleStatus } from '../enums';
|
||||
import { AllowanceScheduleRepository } from '../repositories';
|
||||
|
||||
@Injectable()
|
||||
export class AllowanceService {
|
||||
private readonly logger = new Logger(AllowanceService.name);
|
||||
|
||||
constructor(
|
||||
private readonly allowanceScheduleRepository: AllowanceScheduleRepository,
|
||||
private readonly juniorService: JuniorService,
|
||||
) {}
|
||||
|
||||
async createSchedule(
|
||||
guardianId: string,
|
||||
juniorId: string,
|
||||
body: CreateAllowanceScheduleRequestDto,
|
||||
): Promise<AllowanceSchedule> {
|
||||
const doesBelong = await this.juniorService.doesJuniorBelongToGuardian(guardianId, juniorId);
|
||||
if (!doesBelong) {
|
||||
this.logger.error(`Junior ${juniorId} does not belong to guardian ${guardianId}`);
|
||||
throw new BadRequestException('JUNIOR.NOT_BELONG_TO_GUARDIAN');
|
||||
}
|
||||
|
||||
const existingSchedule = await this.allowanceScheduleRepository.findByGuardianAndJunior(guardianId, juniorId);
|
||||
if (existingSchedule) {
|
||||
throw new BadRequestException('ALLOWANCE.ALREADY_EXISTS');
|
||||
}
|
||||
|
||||
const nextRunAt = this.computeNextRunAt(body.frequency, body.status);
|
||||
return this.allowanceScheduleRepository.createSchedule(guardianId, juniorId, body, nextRunAt);
|
||||
}
|
||||
|
||||
private computeNextRunAt(frequency: AllowanceFrequency, status: AllowanceScheduleStatus): Date {
|
||||
const base = moment();
|
||||
if (status === AllowanceScheduleStatus.OFF) {
|
||||
return base.toDate();
|
||||
}
|
||||
|
||||
switch (frequency) {
|
||||
case AllowanceFrequency.DAILY:
|
||||
return base.add(1, 'day').toDate();
|
||||
case AllowanceFrequency.WEEKLY:
|
||||
return base.add(1, 'week').toDate();
|
||||
case AllowanceFrequency.MONTHLY:
|
||||
return base.add(1, 'month').toDate();
|
||||
default:
|
||||
return base.toDate();
|
||||
}
|
||||
}
|
||||
}
|
||||
3
src/allowance/services/index.ts
Normal file
3
src/allowance/services/index.ts
Normal file
@ -0,0 +1,3 @@
|
||||
export * from './allowance-queue.service';
|
||||
export * from './allowance.service';
|
||||
export * from './allowance-worker.service';
|
||||
@ -9,6 +9,7 @@ import { LoggerModule } from 'nestjs-pino';
|
||||
import { DataSource } from 'typeorm';
|
||||
import { addTransactionalDataSource } from 'typeorm-transactional';
|
||||
import { AuthModule } from './auth/auth.module';
|
||||
import { AllowanceModule } from './allowance/allowance.module';
|
||||
import { CardModule } from './card/card.module';
|
||||
import { CacheModule } from './common/modules/cache/cache.module';
|
||||
import { LookupModule } from './common/modules/lookup/lookup.module';
|
||||
@ -58,6 +59,7 @@ import { MoneyRequestModule } from './money-request/money-request.module';
|
||||
ScheduleModule.forRoot(),
|
||||
// App modules
|
||||
AuthModule,
|
||||
AllowanceModule,
|
||||
UserModule,
|
||||
|
||||
CustomerModule,
|
||||
|
||||
@ -1,8 +1,14 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
import { TypeOrmModule } from '@nestjs/typeorm';
|
||||
import { AllowanceModule } from '~/allowance/allowance.module';
|
||||
import { CronRun } from './entities';
|
||||
import { CronRunRepository } from './repositories';
|
||||
import { BaseCronService } from './services';
|
||||
import { CronRunService } from './services/cron-run.service';
|
||||
import { AllowanceScheduleCron } from './tasks';
|
||||
|
||||
@Module({
|
||||
imports: [],
|
||||
providers: [BaseCronService],
|
||||
imports: [AllowanceModule, TypeOrmModule.forFeature([CronRun])],
|
||||
providers: [BaseCronService, CronRunService, CronRunRepository, AllowanceScheduleCron],
|
||||
})
|
||||
export class CronModule {}
|
||||
|
||||
39
src/cron/entities/cron-run.entity.ts
Normal file
39
src/cron/entities/cron-run.entity.ts
Normal file
@ -0,0 +1,39 @@
|
||||
import {
|
||||
BaseEntity,
|
||||
Column,
|
||||
CreateDateColumn,
|
||||
Entity,
|
||||
PrimaryGeneratedColumn,
|
||||
UpdateDateColumn,
|
||||
} from 'typeorm';
|
||||
import { CronRunStatus } from '../enums/cron-run-status.enum';
|
||||
|
||||
@Entity('cron_runs')
|
||||
export class CronRun extends BaseEntity {
|
||||
@PrimaryGeneratedColumn('uuid')
|
||||
id!: string;
|
||||
|
||||
@Column({ type: 'varchar', name: 'job_name' })
|
||||
jobName!: string;
|
||||
|
||||
@Column({ type: 'varchar', name: 'status', default: CronRunStatus.SUCCESS })
|
||||
status!: CronRunStatus;
|
||||
|
||||
@Column({ type: 'int', name: 'processed_count', default: 0 })
|
||||
processedCount!: number;
|
||||
|
||||
@Column({ type: 'text', name: 'error_message', nullable: true })
|
||||
errorMessage!: string | null;
|
||||
|
||||
@Column({ type: 'timestamp with time zone', name: 'started_at' })
|
||||
startedAt!: Date;
|
||||
|
||||
@Column({ type: 'timestamp with time zone', name: 'finished_at', nullable: true })
|
||||
finishedAt!: Date | null;
|
||||
|
||||
@CreateDateColumn({ type: 'timestamp with time zone', default: () => 'CURRENT_TIMESTAMP', name: 'created_at' })
|
||||
createdAt!: Date;
|
||||
|
||||
@UpdateDateColumn({ type: 'timestamp with time zone', default: () => 'CURRENT_TIMESTAMP', name: 'updated_at' })
|
||||
updatedAt!: Date;
|
||||
}
|
||||
1
src/cron/entities/index.ts
Normal file
1
src/cron/entities/index.ts
Normal file
@ -0,0 +1 @@
|
||||
export * from './cron-run.entity';
|
||||
4
src/cron/enums/cron-run-status.enum.ts
Normal file
4
src/cron/enums/cron-run-status.enum.ts
Normal file
@ -0,0 +1,4 @@
|
||||
export enum CronRunStatus {
|
||||
SUCCESS = 'SUCCESS',
|
||||
FAILED = 'FAILED',
|
||||
}
|
||||
44
src/cron/repositories/cron-run.repository.ts
Normal file
44
src/cron/repositories/cron-run.repository.ts
Normal file
@ -0,0 +1,44 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { Repository } from 'typeorm';
|
||||
import { CronRun } from '../entities';
|
||||
import { CronRunStatus } from '../enums/cron-run-status.enum';
|
||||
|
||||
@Injectable()
|
||||
export class CronRunRepository {
|
||||
constructor(@InjectRepository(CronRun) private readonly cronRunRepository: Repository<CronRun>) {}
|
||||
|
||||
createRun(jobName: string, startedAt: Date): Promise<CronRun> {
|
||||
return this.cronRunRepository.save(
|
||||
this.cronRunRepository.create({
|
||||
jobName,
|
||||
startedAt,
|
||||
status: CronRunStatus.SUCCESS,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
markSuccess(id: string, processedCount: number, finishedAt: Date) {
|
||||
return this.cronRunRepository.update(
|
||||
{ id },
|
||||
{
|
||||
status: CronRunStatus.SUCCESS,
|
||||
processedCount,
|
||||
finishedAt,
|
||||
errorMessage: null,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
markFailure(id: string, processedCount: number, finishedAt: Date, errorMessage: string) {
|
||||
return this.cronRunRepository.update(
|
||||
{ id },
|
||||
{
|
||||
status: CronRunStatus.FAILED,
|
||||
processedCount,
|
||||
finishedAt,
|
||||
errorMessage,
|
||||
},
|
||||
);
|
||||
}
|
||||
}
|
||||
1
src/cron/repositories/index.ts
Normal file
1
src/cron/repositories/index.ts
Normal file
@ -0,0 +1 @@
|
||||
export * from './cron-run.repository';
|
||||
20
src/cron/services/cron-run.service.ts
Normal file
20
src/cron/services/cron-run.service.ts
Normal file
@ -0,0 +1,20 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
import { CronRun } from '../entities';
|
||||
import { CronRunRepository } from '../repositories';
|
||||
|
||||
@Injectable()
|
||||
export class CronRunService {
|
||||
constructor(private readonly cronRunRepository: CronRunRepository) {}
|
||||
|
||||
start(jobName: string): Promise<CronRun> {
|
||||
return this.cronRunRepository.createRun(jobName, new Date());
|
||||
}
|
||||
|
||||
success(id: string, processedCount: number) {
|
||||
return this.cronRunRepository.markSuccess(id, processedCount, new Date());
|
||||
}
|
||||
|
||||
failure(id: string, processedCount: number, errorMessage: string) {
|
||||
return this.cronRunRepository.markFailure(id, processedCount, new Date(), errorMessage);
|
||||
}
|
||||
}
|
||||
@ -1 +1,2 @@
|
||||
export * from './base-cron.service';
|
||||
export * from './cron-run.service';
|
||||
63
src/cron/tasks/allowance-schedule.cron.ts
Normal file
63
src/cron/tasks/allowance-schedule.cron.ts
Normal file
@ -0,0 +1,63 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
import { Cron, CronExpression } from '@nestjs/schedule';
|
||||
import { AllowanceQueueService } from '~/allowance/services';
|
||||
import { AllowanceScheduleRepository } from '~/allowance/repositories';
|
||||
import { BaseCronService } from '../services';
|
||||
import { CronRunService } from '../services/cron-run.service';
|
||||
|
||||
@Injectable()
|
||||
export class AllowanceScheduleCron {
|
||||
private readonly logger = new Logger(AllowanceScheduleCron.name);
|
||||
private readonly lockKey = 'cron:allowance:enqueue';
|
||||
private readonly jobName = 'allowance.schedule.enqueue';
|
||||
|
||||
constructor(
|
||||
private readonly baseCronService: BaseCronService,
|
||||
private readonly allowanceScheduleRepository: AllowanceScheduleRepository,
|
||||
private readonly allowanceQueueService: AllowanceQueueService,
|
||||
private readonly cronRunService: CronRunService,
|
||||
) {}
|
||||
|
||||
@Cron(CronExpression.EVERY_5_MINUTES)
|
||||
async enqueueDueSchedules() {
|
||||
const hasLock = await this.baseCronService.acquireLock(this.lockKey, 240);
|
||||
if (!hasLock) {
|
||||
return;
|
||||
}
|
||||
|
||||
const cronRun = await this.cronRunService.start(this.jobName);
|
||||
let processedCount = 0;
|
||||
try {
|
||||
const batchSize = 100;
|
||||
let cursor: { nextRunAt: Date; id: string } | undefined;
|
||||
let processedBatches = 0;
|
||||
|
||||
while (processedBatches < 50) {
|
||||
const schedules = await this.allowanceScheduleRepository.findDueSchedulesBatch(batchSize, cursor);
|
||||
if (!schedules.length) {
|
||||
break;
|
||||
}
|
||||
|
||||
await Promise.all(
|
||||
schedules.map((schedule) => this.allowanceQueueService.enqueueSchedule(schedule.id, schedule.nextRunAt)),
|
||||
);
|
||||
|
||||
const last = schedules[schedules.length - 1];
|
||||
cursor = { nextRunAt: last.nextRunAt, id: last.id };
|
||||
processedBatches += 1;
|
||||
processedCount += schedules.length;
|
||||
|
||||
if (schedules.length < batchSize) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
await this.cronRunService.success(cronRun.id, processedCount);
|
||||
} catch (error) {
|
||||
const stack = error instanceof Error ? error.stack : undefined;
|
||||
this.logger.error('Failed to enqueue allowance schedules', stack || error);
|
||||
await this.cronRunService.failure(cronRun.id, processedCount, String(stack || error));
|
||||
} finally {
|
||||
await this.baseCronService.releaseLock(this.lockKey);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1 @@
|
||||
export * from './allowance-schedule.cron';
|
||||
|
||||
@ -0,0 +1,24 @@
|
||||
import { MigrationInterface, QueryRunner } from "typeorm";
|
||||
|
||||
export class CreateAllowanceSchedulesAndCredits1769423488963 implements MigrationInterface {
|
||||
name = 'CreateAllowanceSchedulesAndCredits1769423488963'
|
||||
|
||||
public async up(queryRunner: QueryRunner): Promise<void> {
|
||||
await queryRunner.query(`CREATE TABLE "allowance_credits" ("id" uuid NOT NULL DEFAULT uuid_generate_v4(), "schedule_id" uuid NOT NULL, "transaction_id" uuid, "amount" numeric(12,2) NOT NULL, "run_at" TIMESTAMP WITH TIME ZONE NOT NULL, "credited_at" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(), CONSTRAINT "PK_4d8f104f20199b6c23a92e265bf" PRIMARY KEY ("id"))`);
|
||||
await queryRunner.query(`CREATE TABLE "allowance_schedules" ("id" uuid NOT NULL DEFAULT uuid_generate_v4(), "amount" numeric(12,2) NOT NULL, "frequency" character varying NOT NULL, "status" character varying NOT NULL DEFAULT 'ON', "next_run_at" TIMESTAMP WITH TIME ZONE NOT NULL, "last_run_at" TIMESTAMP WITH TIME ZONE, "created_at" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(), "updated_at" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(), "guardian_id" uuid NOT NULL, "junior_id" uuid NOT NULL, CONSTRAINT "PK_27ebe08d13044e8739af557b7a5" PRIMARY KEY ("id"))`);
|
||||
await queryRunner.query(`ALTER TABLE "allowance_credits" ADD CONSTRAINT "FK_9b15567a82f05604001fde8a914" FOREIGN KEY ("schedule_id") REFERENCES "allowance_schedules"("id") ON DELETE CASCADE ON UPDATE NO ACTION`);
|
||||
await queryRunner.query(`ALTER TABLE "allowance_credits" ADD CONSTRAINT "FK_b179c206f7bb6f10b5168e583b8" FOREIGN KEY ("transaction_id") REFERENCES "transactions"("id") ON DELETE SET NULL ON UPDATE NO ACTION`);
|
||||
await queryRunner.query(`ALTER TABLE "allowance_schedules" ADD CONSTRAINT "FK_43eb94744e09d8349811c148351" FOREIGN KEY ("guardian_id") REFERENCES "guardians"("id") ON DELETE CASCADE ON UPDATE NO ACTION`);
|
||||
await queryRunner.query(`ALTER TABLE "allowance_schedules" ADD CONSTRAINT "FK_dd8a608d7f50cb120fa6f7b163f" FOREIGN KEY ("junior_id") REFERENCES "juniors"("id") ON DELETE CASCADE ON UPDATE NO ACTION`);
|
||||
}
|
||||
|
||||
public async down(queryRunner: QueryRunner): Promise<void> {
|
||||
await queryRunner.query(`ALTER TABLE "allowance_schedules" DROP CONSTRAINT "FK_dd8a608d7f50cb120fa6f7b163f"`);
|
||||
await queryRunner.query(`ALTER TABLE "allowance_schedules" DROP CONSTRAINT "FK_43eb94744e09d8349811c148351"`);
|
||||
await queryRunner.query(`ALTER TABLE "allowance_credits" DROP CONSTRAINT "FK_b179c206f7bb6f10b5168e583b8"`);
|
||||
await queryRunner.query(`ALTER TABLE "allowance_credits" DROP CONSTRAINT "FK_9b15567a82f05604001fde8a914"`);
|
||||
await queryRunner.query(`DROP TABLE "allowance_schedules"`);
|
||||
await queryRunner.query(`DROP TABLE "allowance_credits"`);
|
||||
}
|
||||
|
||||
}
|
||||
14
src/db/migrations/1769602683670-CreateCronRunsTable.ts
Normal file
14
src/db/migrations/1769602683670-CreateCronRunsTable.ts
Normal file
@ -0,0 +1,14 @@
|
||||
import { MigrationInterface, QueryRunner } from "typeorm";
|
||||
|
||||
export class CreateCronRunsTable1769602683670 implements MigrationInterface {
|
||||
name = 'CreateCronRunsTable1769602683670'
|
||||
|
||||
public async up(queryRunner: QueryRunner): Promise<void> {
|
||||
await queryRunner.query(`CREATE TABLE "cron_runs" ("id" uuid NOT NULL DEFAULT uuid_generate_v4(), "job_name" character varying NOT NULL, "status" character varying NOT NULL DEFAULT 'SUCCESS', "processed_count" integer NOT NULL DEFAULT '0', "error_message" text, "started_at" TIMESTAMP WITH TIME ZONE NOT NULL, "finished_at" TIMESTAMP WITH TIME ZONE, "created_at" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(), "updated_at" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT now(), CONSTRAINT "PK_43e18ce1778dc9b20a137b84fd5" PRIMARY KEY ("id"))`);
|
||||
}
|
||||
|
||||
public async down(queryRunner: QueryRunner): Promise<void> {
|
||||
await queryRunner.query(`DROP TABLE "cron_runs"`);
|
||||
}
|
||||
|
||||
}
|
||||
@ -11,4 +11,5 @@ export * from './1765804942393-AddKycFieldsAndTransactions';
|
||||
export * from './1765877128065-AddNationalIdToKycTransactions';
|
||||
export * from './1765891028260-RemoveOldCustomerColumns';
|
||||
export * from './1765975126402-RemoveAddressColumns';
|
||||
export * from './1768395622276-AddTimezoneFields';
|
||||
export * from './1768395622276-AddTimezoneFields';export * from './1769423488963-CreateAllowanceSchedulesAndCredits';
|
||||
export * from './1769602683670-CreateCronRunsTable';
|
||||
|
||||
@ -11,6 +11,7 @@ import {
|
||||
} from 'typeorm';
|
||||
import { Customer } from '~/customer/entities';
|
||||
import { Junior } from '~/junior/entities';
|
||||
import { AllowanceSchedule } from '~/allowance/entities/allowance-schedule.entity';
|
||||
import { MoneyRequest } from '~/money-request/entities/money-request.entity';
|
||||
|
||||
@Entity('guardians')
|
||||
@ -31,6 +32,9 @@ export class Guardian extends BaseEntity {
|
||||
@OneToMany(() => MoneyRequest, (moneyRequest) => moneyRequest.guardian)
|
||||
moneyRequests!: MoneyRequest[];
|
||||
|
||||
@OneToMany(() => AllowanceSchedule, (schedule) => schedule.guardian)
|
||||
allowanceSchedules!: AllowanceSchedule[];
|
||||
|
||||
@CreateDateColumn({ name: 'created_at', type: 'timestamp with time zone', default: () => 'CURRENT_TIMESTAMP' })
|
||||
createdAt!: Date;
|
||||
|
||||
|
||||
@ -1,4 +1,5 @@
|
||||
import { Controller, Get } from '@nestjs/common';
|
||||
import { ConfigService } from '@nestjs/config';
|
||||
import { ApiTags } from '@nestjs/swagger';
|
||||
import { HealthCheck, HealthCheckService, HttpHealthIndicator, TypeOrmHealthIndicator } from '@nestjs/terminus';
|
||||
import { SkipThrottle } from '@nestjs/throttler';
|
||||
@ -11,6 +12,7 @@ export class HealthController {
|
||||
private readonly healthCheckService: HealthCheckService,
|
||||
private readonly databaseHealthIndicator: TypeOrmHealthIndicator,
|
||||
private readonly httpHealthIndicator: HttpHealthIndicator,
|
||||
private readonly configService: ConfigService,
|
||||
) {}
|
||||
|
||||
@Get()
|
||||
@ -24,9 +26,23 @@ export class HealthController {
|
||||
checkHealthDetails() {
|
||||
const healthIndicators = [
|
||||
() => this.databaseHealthIndicator.pingCheck('database'),
|
||||
// add your own health indicators here
|
||||
];
|
||||
|
||||
const rabbitManagementUrl = this.configService.get<string>('RABBITMQ_MANAGEMENT_URL');
|
||||
if (rabbitManagementUrl) {
|
||||
const user = this.configService.get<string>('RABBITMQ_MANAGEMENT_USER');
|
||||
const pass = this.configService.get<string>('RABBITMQ_MANAGEMENT_PASS');
|
||||
const headers: Record<string, string> = {};
|
||||
if (user && pass) {
|
||||
const encoded = Buffer.from(`${user}:${pass}`).toString('base64');
|
||||
headers.Authorization = `Basic ${encoded}`;
|
||||
}
|
||||
|
||||
healthIndicators.push(() =>
|
||||
this.httpHealthIndicator.pingCheck('rabbitmq', rabbitManagementUrl, { headers }),
|
||||
);
|
||||
}
|
||||
|
||||
return this.healthCheckService.check(healthIndicators);
|
||||
}
|
||||
}
|
||||
|
||||
@ -41,6 +41,7 @@
|
||||
"ALLOWANCE": {
|
||||
"START_DATE_BEFORE_TODAY": "لا يمكن أن يكون تاريخ البدء قبل اليوم.",
|
||||
"START_DATE_AFTER_END_DATE": "لا يمكن أن يكون تاريخ البدء بعد تاريخ النهاية.",
|
||||
"ALREADY_EXISTS": "يوجد جدول مصروف قائم لهذا الطفل بالفعل.",
|
||||
"NOT_FOUND": "لم يتم العثور على المصروف.",
|
||||
"DOES_NOT_BELONG_TO_JUNIOR": "المصروف لا يخص الطفل."
|
||||
},
|
||||
|
||||
@ -94,10 +94,12 @@
|
||||
"name": "الاسم",
|
||||
"amount": "المبلغ",
|
||||
"type": "النوع",
|
||||
"frequency": "التكرار",
|
||||
"startDate": "تاريخ البدء",
|
||||
"endDate": "تاريخ النهاية",
|
||||
"numberOfTransactions": "عدد المعاملات",
|
||||
"juniorId": "معرّف الطفل"
|
||||
"juniorId": "معرّف الطفل",
|
||||
"status": "الحالة"
|
||||
},
|
||||
"allowanceChangeRequest": {
|
||||
"reason": "السبب",
|
||||
|
||||
@ -40,6 +40,7 @@
|
||||
"ALLOWANCE": {
|
||||
"START_DATE_BEFORE_TODAY": "The start date cannot be before today.",
|
||||
"START_DATE_AFTER_END_DATE": "The start date cannot be after the end date.",
|
||||
"ALREADY_EXISTS": "An allowance schedule already exists for this child.",
|
||||
"NOT_FOUND": "The allowance was not found.",
|
||||
"DOES_NOT_BELONG_TO_JUNIOR": "The allowance does not belong to the junior."
|
||||
},
|
||||
|
||||
@ -98,10 +98,12 @@
|
||||
"name": "Name",
|
||||
"amount": "Amount",
|
||||
"type": "Type",
|
||||
"frequency": "Frequency",
|
||||
"startDate": "Start date",
|
||||
"endDate": "End date",
|
||||
"numberOfTransactions": "Number of transactions",
|
||||
"juniorId": "Junior ID"
|
||||
"juniorId": "Junior ID",
|
||||
"status": "Status"
|
||||
},
|
||||
"allowanceChangeRequest": {
|
||||
"reason": "Reason",
|
||||
|
||||
@ -11,6 +11,7 @@ import {
|
||||
PrimaryColumn,
|
||||
UpdateDateColumn,
|
||||
} from 'typeorm';
|
||||
import { AllowanceSchedule } from '~/allowance/entities/allowance-schedule.entity';
|
||||
import { Customer } from '~/customer/entities';
|
||||
import { Guardian } from '~/guardian/entities/guradian.entity';
|
||||
import { MoneyRequest } from '~/money-request/entities/money-request.entity';
|
||||
@ -45,6 +46,9 @@ export class Junior extends BaseEntity {
|
||||
@OneToMany(() => MoneyRequest, (moneyRequest) => moneyRequest.junior)
|
||||
moneyRequests!: MoneyRequest[];
|
||||
|
||||
@OneToMany(() => AllowanceSchedule, (schedule) => schedule.junior)
|
||||
allowanceSchedules!: AllowanceSchedule[];
|
||||
|
||||
@CreateDateColumn({ name: 'created_at', type: 'timestamp with time zone', default: () => 'CURRENT_TIMESTAMP' })
|
||||
createdAt!: Date;
|
||||
|
||||
|
||||
1
src/types/amqplib.d.ts
vendored
Normal file
1
src/types/amqplib.d.ts
vendored
Normal file
@ -0,0 +1 @@
|
||||
declare module 'amqplib';
|
||||
Reference in New Issue
Block a user