App Code Challenge Jorge Paredes

anti-fraud-service/kafka/consumer.js

@@ -0,0 +1,29 @@
+const { Kafka } = require('kafkajs');
+const kafka = new Kafka({ clientId: 'anti-fraud-service', brokers: ['localhost:9092'] });
+const consumer = kafka.consumer({ groupId: 'anti-fraud-group' });
+const axios = require('axios');
+
+const startConsumer = async () => {
+  await consumer.connect();
+  await consumer.subscribe({ topic: 'transaction-validation', fromBeginning: true });
+
+  await consumer.run({
+    eachMessage: async ({ message }) => {
+      const transaction = JSON.parse(message.value.toString());
+      const status = transaction.amount > 1000 ? 'rejected' : 'approved';
+
+      await updateTransactionStatus({ id: transaction.id, status });
+    },
+  });
+};
+
+const updateTransactionStatus = async(transaction) => {
+  try {
+    const response = await axios.patch('http://localhost:3000/transactions/update-status', transaction);
+    console.log('Transaction status updated:', response.data);
+  } catch (error) {
+    console.error('Error updating transaction status:', error);
+  }
+}
+
+module.exports = startConsumer;

anti-fraud-service/package.json

@@ -0,0 +1,16 @@
+{
+  "name": "anti-fraud-service",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "axios": "^1.7.7",
+    "kafkajs": "^2.2.4"
+  }
+}

anti-fraud-service/server.js

@@ -0,0 +1,5 @@
+const startConsumer = require('./kafka/consumer');
+
+startConsumer()
+  .then(() => console.log('Anti-Fraud Service is listening for transactions'))
+  .catch(err => console.log('Error starting Anti-Fraud Service:', err));

docker-compose.yml

@@ -3,10 +3,11 @@ services:
   postgres:
     image: postgres:14
     ports:
-      - "5432:5432"
+      - "5433:5432"
     environment:
-      - POSTGRES_USER=postgres
-      - POSTGRES_PASSWORD=postgres
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_DB: app-code-challenge
   zookeeper:
     image: confluentinc/cp-zookeeper:5.5.3
     environment:
@@ -22,4 +23,4 @@ services:
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
       KAFKA_JMX_PORT: 9991
     ports:
-      - 9092:9092
+      - 9092:9092

transaction-service/config/config.js

@@ -0,0 +1,10 @@
+require('dotenv').config();
+
+module.exports = {
+  username: 'postgres',
+  password: 'postgres',
+  database: 'app-code-challenge',
+  host: 'localhost',
+  dialect: 'postgres',
+  port: 5433
+};

transaction-service/controllers/transactionController.js

@@ -0,0 +1,34 @@
+const sequelize = require('../database/database'); 
+const Transaction = require('../models/transaction')(sequelize);
+const sendMessage = require('../kafka/producer');
+
+exports.createTransaction = async (req, res) => {
+  try {
+    const { amount } = req.body;
+    const transaction = await Transaction.create({ amount });
+
+    // Send to anti-fraud validation
+    await sendMessage({ id: transaction.id, amount });
+
+    res.status(201).json(transaction);
+  } catch (error) {
+    res.status(500).json({ error: 'Internal Server Error' });
+  }
+};
+
+exports.updateTransactionStatus = async (req, res) => {
+  try {
+    const { id, status } = req.body;
+    const transaction = await Transaction.findByPk(id);
+
+    if (transaction) {
+      transaction.status = status;
+      await transaction.save();
+      res.status(200).json(transaction);
+    } else {
+      res.status(404).json({ error: 'Transaction not found' });
+    }
+  } catch (error) {
+    res.status(500).json({ error: 'Internal Server Error' });
+  }
+};

transaction-service/database/database.js

@@ -0,0 +1,11 @@
+const { Sequelize } = require('sequelize');
+const config = require('../config/config');
+
+// Setup PostgreSQL database connection (adjust with your actual DB connection details)
+const sequelize = new Sequelize(config.database,config.username,config.password,{
+  dialect: 'postgres',
+  host: config.host, // Use 'localhost' if you're running the DB locally
+  port: config.port,        // Adjust the port if needed (e.g., 5433 for Docker)
+});
+
+module.exports = sequelize;

transaction-service/kafka/producer.js

@@ -0,0 +1,15 @@
+const { Kafka } = require('kafkajs');
+const kafka = new Kafka({ clientId: 'transaction-service', brokers: ['localhost:9092'] });
+
+const producer = kafka.producer();
+
+const sendMessage = async (message) => {
+  await producer.connect();
+  await producer.send({
+    topic: 'transaction-validation',
+    messages: [{ value: JSON.stringify(message) }],
+  });
+  await producer.disconnect();
+};
+
+module.exports = sendMessage;

transaction-service/migrations/initDatabase.js

@@ -0,0 +1,13 @@
+const sequelize = require('../database/database');  
+const config = require('../config/config');
+
+(async () => {
+  try {
+    await sequelize.query(`CREATE DATABASE ${config.database}`);
+    console.log('Database created successfully!');
+  } catch (error) {
+    console.error('Error creating database:', error);
+  } finally {
+    await sequelize.close();
+  }
+})();

transaction-service/migrations/syncDatabase.js

@@ -0,0 +1,11 @@
+const sequelize = require('../database/database');  // Import your Sequelize instance
+const Transaction = require('../models/Transaction')(sequelize);  // Import your Transaction model
+
+// Sync the models to the database
+sequelize.sync({ force: false })  // Set force: true to drop and recreate the table
+  .then(() => {
+    console.log('Database synced and table created!');
+  })
+  .catch((error) => {
+    console.error('Error syncing database:', error);
+  });

transaction-service/models/transaction.js

@@ -0,0 +1,13 @@
+const { Model, DataTypes } = require('sequelize');
+
+module.exports = (sequelize) => {
+  class Transaction extends Model {}
+
+  Transaction.init({
+    id: { type: DataTypes.UUID, primaryKey: true, defaultValue: DataTypes.UUIDV4 },
+    amount: { type: DataTypes.DECIMAL, allowNull: false },
+    status: { type: DataTypes.ENUM('pending', 'approved', 'rejected'), defaultValue: 'pending' },
+  }, { sequelize, modelName: 'Transaction' });
+
+  return Transaction;
+};