Merge pull request #1 from sebastiansj70/feature/create-transaction-m…

…odule-v1

Feature/create transaction module v1

apps/transactions/src/adapters/controllers/transaction.controller.ts

@@ -0,0 +1,66 @@
+import { Request, Response } from 'express';
+import { UpdateTransactionUseCase } from '../../application/updateTransaction.usecase';
+import { TransactionRepository } from '../../domain/repositories/transaction.repository';
+import { CreateTransactionUseCase } from '../../application/createTransaction.usecase';
+import { CreateTransactionDTO } from '../../domain/dtos/createTransaction.dto';
+import { validateCreateTransactionDto, validateUpdateTransactionBody } from '../validation/transaction.validator';
+import { sendTransactionMessage } from '../../infrastructure/messageBroker/kafkaProducer';
+import { mapTransactionToKafkaMessage } from '../../infrastructure/mappers/transaction.mapper';
+import { UpdatedData } from '../../domain/dtos/updatedTransaction.dto';
+
+
+export class TransactionController {
+    private transactionRepository: TransactionRepository;
+
+    constructor(transactionRepository: TransactionRepository) {
+        this.transactionRepository = transactionRepository
+    }
+
+
+    public createTransaction = async (req: Request, res: Response) => {
+        try {
+
+            if (!validateCreateTransactionDto(req.body)) {
+                return res.status(400).json({ message: 'incorrect body' });
+            }
+
+            const { accountExternalIdDebit, accountExternalIdCredit, transferTypeId, value } = req.body;
+
+            const createTransactionDto: CreateTransactionDTO = {
+                accountExternalIdDebit,
+                accountExternalIdCredit,
+                transferTypeId,
+                value
+            };
+
+            const createTransactionUseCase = new CreateTransactionUseCase(this.transactionRepository)
+            const savedTransaction = await createTransactionUseCase.execute(createTransactionDto);
+
+            const message = mapTransactionToKafkaMessage(savedTransaction)
+            await sendTransactionMessage(message);
+
+            return res.status(200).json(savedTransaction);
+        } catch (error) {
+            console.error('Error creating transaction:', error);
+            return res.status(500).json({ message: 'Error creating transaction', error });
+        }
+    };
+
+    public updateTransaction = async (data: UpdatedData): Promise<void> => {
+        try {
+            if (!validateUpdateTransactionBody(data)) {
+                throw new Error('incorrect body')
+            }
+
+            const id = data.transactionExternalId
+
+            const updateTransactionUseCase = new UpdateTransactionUseCase(this.transactionRepository);
+            await updateTransactionUseCase.execute(id, data);
+
+        } catch (error: any) {
+            console.error('Error updating transaction:', error);
+            throw error;
+        }
+    };
+
+}

apps/transactions/src/adapters/routes/transaction.routes.ts

@@ -0,0 +1,15 @@
+import { Router, Request, Response } from 'express';
+import { TransactionController } from '../controllers/transaction.controller';
+import { TransactionRepositoryImpl } from '../../domain/repositories/transaction.repository';
+
+const router = Router();
+const transactionRepository = new TransactionRepositoryImpl();
+const transactionController = new TransactionController(transactionRepository)
+
+
+router.post('/create', async (req: Request, res: Response) => {
+    await transactionController.createTransaction(req, res)
+});
+
+
+export default router;

apps/transactions/src/adapters/validation/transaction.validator.ts

@@ -0,0 +1,38 @@
+import { CreateTransactionDTO } from '../../domain/dtos/createTransaction.dto';
+import { UpdatedData } from '../../domain/dtos/updatedTransaction.dto';
+
+export const validateCreateTransactionDto = (data: any): data is CreateTransactionDTO => {
+    const { accountExternalIdDebit, accountExternalIdCredit, transferTypeId, value } = data;
+
+    if (
+        typeof accountExternalIdDebit !== 'string' ||
+        typeof accountExternalIdCredit !== 'string' ||
+        typeof transferTypeId !== 'number' ||
+        ![1, 2, 3].includes(transferTypeId) ||
+        typeof value !== 'number' ||
+        value <= 0
+    ) {
+        return false;
+    }
+
+    return true;
+};
+
+export const validateUpdateTransactionBody = (data: any): data is UpdatedData => {
+    const { transactionExternalId, transactionType, transactionStatus, value, createdAt } = data;
+
+    if (
+        typeof transactionExternalId !== 'string' ||
+        typeof transactionType !== 'object' ||
+        typeof transactionType.name !== 'number' ||
+        typeof transactionStatus !== 'object' ||
+        typeof transactionStatus.name !== 'string' ||
+        typeof value !== 'number' ||
+        value <= 0 ||
+        typeof createdAt !== 'string'
+    ) {
+        return false;
+    }
+
+    return true;
+}

apps/transactions/src/application/createTransaction.usecase.ts

@@ -0,0 +1,21 @@
+import { CreateTransactionDTO } from '../domain/dtos/createTransaction.dto';
+import { Transaction } from '../domain/entities/transaction.entity';
+import { TransactionRepository } from '../domain/repositories/transaction.repository';
+
+export class CreateTransactionUseCase {
+    constructor(private transactionRepository: TransactionRepository) { }
+
+    async execute(dto: CreateTransactionDTO): Promise<Transaction> {
+        const transaction = new Transaction(
+            dto.accountExternalIdDebit,
+            dto.accountExternalIdCredit,
+            dto.transferTypeId,
+            dto.value,
+            'pending',
+            new Date()
+        );
+        const savedTransaction = await this.transactionRepository.save(transaction);
+
+        return savedTransaction;
+    }
+}

apps/transactions/src/application/updateTransaction.usecase.ts

@@ -0,0 +1,19 @@
+import { UpdatedData } from "../domain/dtos/updatedTransaction.dto";
+import { Transaction } from "../domain/entities/transaction.entity";
+import { TransactionRepository } from '../domain/repositories/transaction.repository';
+
+
+export class UpdateTransactionUseCase {
+    constructor(private transactionRepository: TransactionRepository) { }
+
+    async execute(id: string, updatedData: UpdatedData): Promise<Transaction> {
+        const transaction = await this.transactionRepository.findById(id)
+        if (!transaction) {
+            throw new Error('Transaction not found')
+        }
+        transaction.status = updatedData.transactionStatus.name
+        transaction.transferTypeId = updatedData.transactionType.name
+        await this.transactionRepository.update(id, { status: transaction.status, transferTypeId: transaction.transferTypeId })
+        return transaction
+    }
+}

apps/transactions/src/domain/dtos/createTransaction.dto.ts

@@ -0,0 +1,18 @@
+export class CreateTransactionDTO {
+    accountExternalIdDebit: string;
+    accountExternalIdCredit: string;
+    transferTypeId: number;
+    value: number;
+
+    constructor(
+        accountExternalIdDebit: string,
+        accountExternalIdCredit: string,
+        transferTypeId: number,
+        value: number
+    ) {
+        this.accountExternalIdDebit = accountExternalIdDebit;
+        this.accountExternalIdCredit = accountExternalIdCredit;
+        this.transferTypeId = transferTypeId;
+        this.value = value;
+    }
+}

apps/transactions/src/domain/dtos/kafkaTransactionMessage.dto.ts

@@ -0,0 +1,11 @@
+export interface KafkaTransactionMessage {
+    transactionExternalId: string;
+    transactionType: {
+        name: number;
+    };
+    transactionStatus: {
+        name: string;
+    };
+    value: number;
+    createdAt: Date;
+}

apps/transactions/src/domain/dtos/updatedTransaction.dto.ts

@@ -0,0 +1,11 @@
+export interface UpdatedData {
+    transactionExternalId: string;
+    transactionType: {
+        name: number;
+    };
+    transactionStatus: {
+        name: string;
+    };
+    value: number;
+    createdAt: string;
+}

apps/transactions/src/domain/entities/transaction.entity.ts

@@ -0,0 +1,42 @@
+import { Entity, PrimaryGeneratedColumn, Column } from 'typeorm';
+const isSQLite = process.env.NODE_ENV === 'test'
+
+@Entity('transactions')
+export class Transaction {
+    @PrimaryGeneratedColumn()
+    id!: string;
+
+    @Column()
+    accountExternalIdDebit: string;
+
+    @Column()
+    accountExternalIdCredit: string;
+
+    @Column()
+    transferTypeId: number;
+
+    @Column()
+    value: number;
+
+    @Column({ default: 'pending' })
+    status: string;
+
+    @Column({ type: isSQLite ? 'datetime' : 'timestamp', default: () => 'CURRENT_TIMESTAMP' })
+    createdAt: Date;
+
+    constructor(
+        accountExternalIdDebit: string,
+        accountExternalIdCredit: string,
+        transferTypeId: number,
+        value: number,
+        transactionStatus?: string,
+        createdAt?: Date
+    ) {
+        this.accountExternalIdDebit = accountExternalIdDebit;
+        this.accountExternalIdCredit = accountExternalIdCredit;
+        this.transferTypeId = transferTypeId;
+        this.value = value;
+        this.status = transactionStatus || 'pending';
+        this.createdAt = createdAt || new Date();
+    }
+}

apps/transactions/src/domain/repositories/transaction.repository.ts

@@ -0,0 +1,37 @@
+
+import { Repository, UpdateResult } from 'typeorm';
+import { Transaction } from '../entities/transaction.entity';
+import AppDataSource from '../../infrastructure/database/transaction.ormconfig';
+
+
+export interface TransactionRepository {
+    save(transaction: Transaction): Promise<Transaction>;
+    findById(id: string): Promise<Transaction | null>;
+    findAll(): Promise<Transaction[]>;
+    update(id: string, updatedTransaction: Partial<Transaction>): Promise<UpdateResult>
+}
+
+export class TransactionRepositoryImpl implements TransactionRepository {
+    private readonly repository: Repository<Transaction>;
+
+    constructor() {
+        this.repository = AppDataSource.getRepository(Transaction);
+    }
+
+    async save(transaction: Transaction): Promise<Transaction> {
+        return await this.repository.save(transaction);
+    }
+
+    async findById(id: string): Promise<Transaction | null> {
+        return await this.repository.findOne({ where: { id } }) || null;
+    }
+
+    async findAll(): Promise<Transaction[]> {
+        return await this.repository.find();
+    }
+
+    async update(id: string, updatedTransaction: Partial<Transaction>): Promise<UpdateResult> {
+        return await this.repository.update(id, updatedTransaction);
+
+    }
+}

apps/transactions/src/infrastructure/database/transaction.ormconfig.ts

@@ -0,0 +1,30 @@
+import { DataSource } from 'typeorm';
+import { Transaction } from '../../domain/entities/transaction.entity';
+import dotenv from 'dotenv';
+
+let AppDataSource: DataSource
+if (process.env.NODE_ENV === 'test') {
+    dotenv.config({ path: '.env.test' });
+    AppDataSource = new DataSource({
+        type: 'sqlite',
+        database: './db.sqlite',
+        synchronize: true,
+        logging: ['query', 'error'],
+        entities: [Transaction],
+    });
+} else {
+    dotenv.config();
+    AppDataSource = new DataSource({
+        type: process.env.DB_TYPE as 'postgres',
+        host: process.env.DB_HOST,
+        port: parseInt(process.env.DB_PORT || '5432', 10),
+        username: process.env.DB_USERNAME,
+        password: process.env.DB_PASSWORD,
+        database: process.env.DB_DATABASE,
+        synchronize: Boolean(process.env.DB_SYNCHRONIZE),
+        logging: Boolean(process.env.DB_LOGGING),
+        entities: [Transaction],
+    });
+}
+
+export default AppDataSource;

apps/transactions/src/infrastructure/kafka/kafkaConsumer.ts

@@ -0,0 +1,34 @@
+import { Kafka } from 'kafkajs';
+import dotenv from 'dotenv';
+import { TransactionRepositoryImpl } from '../../domain/repositories/transaction.repository';
+import { TransactionController } from '../../adapters/controllers/transaction.controller';
+
+dotenv.config();
+
+const kafka = new Kafka({
+    clientId: process.env.KAFKA_CLIENT_ID,
+    brokers: [process.env.KAFKA_BROKER || 'localhost:9092'],
+});
+
+const consumer = kafka.consumer({ groupId: process.env.KAFKA_GROUP_ID as string });
+const transactionRepository = new TransactionRepositoryImpl();
+const transactionController = new TransactionController(transactionRepository)
+
+export const consumeTransactionMessages = async () => {
+    await consumer.connect();
+    await consumer.subscribe({ topic: process.env.KAFKA_TOPIC_UPDATE || 'antifraud-transactions-status', fromBeginning: true });
+
+    await consumer.run({
+        eachMessage: async ({ topic, partition, message }) => {
+            const transactionData = JSON.parse(message.value?.toString() || '{}');
+            console.log(`Mensaje recibido en el tópico ${topic} - Partición ${partition}:`, transactionData);
+
+            try {
+                await transactionController.updateTransaction(transactionData);
+                console.log("mensaje procesado exitosamente");
+            } catch (error) {
+                console.error(`Error al procesar la transacción en el tópico ${topic} - message ${message}: `, error);
+            }
+        },
+    });
+};

apps/transactions/src/infrastructure/mappers/transaction.mapper.ts

@@ -0,0 +1,16 @@
+import { Transaction } from "../../domain/entities/transaction.entity";
+import { KafkaTransactionMessage } from "../../domain/dtos/kafkaTransactionMessage.dto";
+
+export const mapTransactionToKafkaMessage = (transaction: Transaction): KafkaTransactionMessage => {
+    return {
+        transactionExternalId: transaction.id,
+        transactionType: {
+            name: transaction.transferTypeId,
+        },
+        transactionStatus: {
+            name: transaction.status,
+        },
+        value: transaction.value,
+        createdAt: transaction.createdAt,
+    };
+};