DEV Community: Igor Dubinin

Microservices for beginners. Common code. Typescript.

Igor Dubinin — Tue, 28 Nov 2023 19:09:23 +0000

Whole series:

Microservices for beginners

Microservices for beginners. Front-end service. Vue js. Socket.io.

Microservices for beginners. Api Gateway service. Nest js. Kafka.

Microservices for beginners. User service. Nest js. Mongodb. Kafka.

Microservices for beginners. Message service. Nest js. Mongodb. Kafka.

Microservices for beginners. Spam service. Python. Scikit-learn. Kafka.

Microservices for beginners. Toxic service. Python. Tensorflow. Kafka.

Microservices for beginners. Common code. Typescript.

Full code - link

In previous articles I described how to create several services and how to work with them. But sometimes microservices in one project contain a lot of similar code, and I want to reuse it. For example I use a common type for user entities in the api gateway and in the user service - WebUserDto.

export class WebUserDto {
  @ApiProperty({ description: 'Id of user', nullable: false })
  id: string;

  @ApiProperty({ description: 'Login of user', nullable: false })
  login: string;

  @ApiProperty({ description: 'Email of user', nullable: false })
  email: string;

  @ApiProperty({ description: 'Flag of user active', nullable: false })
  active: boolean;

  @ApiProperty({ description: 'Date of user creation', nullable: false })
  created_at: Date;
}

So I made a new project in Github and put all the common DTO inside and built a typescript library.

Installation requirements

Installation

npm install typescript
npm install --save @nestjs/swagger

Initiation tsconfig.json

npx tsc --init

Building

npx tsc

Right now I can use common code as npm library in other projects, in package.json - "micro-dto": "github:Igorok/micro-dto#main".

Microservices for beginners. Toxic service. Python. Tensorflow. Kafka.

Igor Dubinin — Sun, 26 Nov 2023 12:55:12 +0000

Whole series:

Microservices for beginners

Microservices for beginners. Front-end service. Vue js. Socket.io.

Microservices for beginners. Api Gateway service. Nest js. Kafka.

Microservices for beginners. User service. Nest js. Mongodb. Kafka.

Microservices for beginners. Message service. Nest js. Mongodb. Kafka.

Microservices for beginners. Spam service. Python. Scikit-learn. Kafka.

Microservices for beginners. Toxic service. Python. Tensorflow. Kafka.

Toxic service provides analysis of toxic and obscene messages. I use a publish-subscribe pattern through Kafka message broker. And Recurrent Neural Network (RNN) model for text analysis.

Full code - link

Whole scheme:

Short description:

User opens the front-end application in the browser and writes messages.
Front-end service emits messages to the api gateway through socket.io.
Api gateway emits messages in the Kafka topic for new messages.
Message service subscribes to the topic with new messages, saves them and publishes events into the topic for saved messages.
Users in the front-end service receive messages in the browser from the Api gateway.
Toxic service subscribes to the Kafka topic with saved messages. Tensorflow model analyzes messages. Messages which were predicted as toxic do publish into the Kafka topic for analyzed messages.
Message and User services listen to events with spam and toxic messages, mark them in the database, and apply domain logic after receiving these events.

Scheme of toxic service:

src/index.py - this file contains initialization of the service. src/index.py - this file contains initialization of the service. main function gets configuration, connects to Kafka consumer and producer, and provides publish-subscribe events.

from infrastructure.config import getConfig
from toxic_module.toxic_service import ToxicService
from infrastructure.kafka import getKafkaConsumer, getKafkaProducer


def main():
    config = getConfig()
    consumer = getKafkaConsumer(config)
    producer = getKafkaProducer(config)
    toxic_service = ToxicService()

    print('toxic-service started')

    for message in consumer:
        try:
            if not message.value.get('message'):
                continue

            prediction = toxic_service.is_toxic(message.value.get('message'))
            if prediction['is_toxic']:
                response = {
                    'id': message.value.get('id'),
                    'user_id': message.value.get('user_id'),
                    'analysis': prediction['prediction']
                }
                producer.send(config['KAFKA_ANALYSIS_MESSAGE_TOPIC'], response)

        except Exception as error:
            print('error', error)


if __name__ == '__main__':
    main()

src/infrastructure/config.py - this is the infrastructure helper, provides configuration for service.

import os
from dotenv import dotenv_values


def getConfig():
    config = {
        **dotenv_values('./config/.env.dev'),  # load shared development variables
        **dotenv_values('./config/.env.prod'),  # load sensitive variables
        **os.environ,  # override loaded values with environment variables
    }
    return {
        'KAFKA_URI': config['KAFKA_URI'],
        'KAFKA_READY_MESSAGE_TOPIC': config.get('KAFKA_READY_MESSAGE_TOPIC'),
        'KAFKA_READY_MESSAGE_GROUP': config.get('KAFKA_READY_MESSAGE_GROUP'),
        'KAFKA_ANALYSIS_MESSAGE_TOPIC': config.get('KAFKA_ANALYSIS_MESSAGE_TOPIC'),
        'KAFKA_ANALYSIS_MESSAGE_GROUP': config.get('KAFKA_ANALYSIS_MESSAGE_GROUP'),
    }

src/infrastructure/kafka.py -this is the infrastructure helper, provides connection to Kafka.

import json
from kafka import KafkaConsumer
from kafka import KafkaProducer


def getKafkaConsumer(config):
    return KafkaConsumer(
        config['KAFKA_READY_MESSAGE_TOPIC'],
        group_id=config['KAFKA_READY_MESSAGE_GROUP'],
        bootstrap_servers=[config['KAFKA_URI']],
        value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    )

def getKafkaProducer(config):
    return KafkaProducer(
        bootstrap_servers=[config['KAFKA_URI']],
        value_serializer=lambda v: json.dumps(v).encode('utf-8')
    )

src/toxic_module/toxic_service.py - this file provides application logic. Here located functionality for message prediction by neural network.

import os
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
from tensorflow.keras.models import load_model

class ToxicService:
    empty_message = 'Empty message!'
    model_path = './src/toxic_module/toxic_model_lstm'

    def __init__(self):
        self.model = load_model(self.model_path)

    def format(self, prediction):
        keys = ['toxic','severe_toxic', 'obscene', 'threat', 'insult', 'identity_hate']
        return dict(zip(keys, prediction))

    def is_toxic(self, message):
        [prediction] = self.model.predict([message])
        prediction = [bool(round(value)) for value in prediction]

        return {
            'is_toxic': any(prediction),
            'prediction': self.format(prediction),
        }

Model

Model for multi-label prediction of toxic comments, these may be toxic, severe_toxic, obscene, threat, insult, identity_hate.

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from tensorflow.keras import utils
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Embedding, MaxPooling1D, Dropout, LSTM, Bidirectional, SpatialDropout1D, TextVectorization
from tensorflow.keras.callbacks import ModelCheckpoint
from tensorflow.keras.models import load_model

Train data contain text and labels with answers.

train_data = pd.read_csv('/kaggle/input/jigsaw-toxic-comment-classification-challenge/train.csv',index_col='id')
train_data.head()

We need to found which length of message we will analyse for prediction

train_data['comment_length'] = train_data['comment_text'].apply(lambda row: len(row))
train_data.head()

import seaborn as sns
sns.displot(
    data=train_data,
    x="comment_length",
    hue='toxic',
    multiple="stack",
)

Length of toxic comments: max 5000, min 8, mean 295.24604420034, median 123.0, 75% 271.0.

print(
    'max', toxic_data['comment_length'].max(),
    'min', toxic_data['comment_length'].min(),
    'mean', toxic_data['comment_length'].mean(),
    'median', toxic_data['comment_length'].median(),
    '75%', toxic_data['comment_length'].quantile(0.75),
)

So 300 symbols should be enough.

max_comment_len = 300

Test data contain two files with text and labels separately.

test_data = pd.read_csv('/kaggle/input/jigsaw-toxic-comment-classification-challenge/test.csv',index_col='id')
test_data.head()

test_labels_data = pd.read_csv('/kaggle/input/jigsaw-toxic-comment-classification-challenge/test_labels.csv',index_col='id')
test_labels_data.head()

And some labels has value of -1, that indicates it was not used for scoring; (Note: file added after competition close!)

test_data = test_data.join(test_labels_data)
test_data = test_data[test_data['toxic'] != -1]
test_data.head()

I used solution from "Toxic Comment Classification Challenge" on base of RNN, which takes 3rd place. My notebook here.

Neural networks this is a series of algorithms that seek to identify relationships in a data without being explicitly programmed.

Neural networks works with numbers, so we need convert messages to numbers.

TextVectorization - A preprocessing layer which maps text features to integer sequences.

max_tokens - Maximum size of the vocabulary for this layer.

adapt - Computes a vocabulary of string terms from tokens in a dataset.

encoder - this is vocabulary created on base of training messages, this is first layer of network that convert coming messages into list of numbers.

Embeding - turns positive integers (indexes) into dense vectors of fixed size. Dense vectors this is modern kind of representing words as vector, this solution have famose effect: King - Man + Woman = Queen.

SpatialDropout1D - dropout, however, it drops entire 1D feature maps instead of individual elements. Dropout removes some neurons from network, this solution help to avoid overfitting.

LSTM - Long Short-Term Memory layer, this is kind of Recurrent Neural Network. return_sequences - whether to return the last output in the output sequence, or the full sequence.

Dense - densely-connected NN layer. Final layer that contain 6 neurons with result.

num_words = 10000
encoder = TextVectorization(max_tokens=num_words)
encoder.adapt(train_data['comment_text'].values)

model_lstm = Sequential([
    encoder,
    Embedding(
        input_dim=len(encoder.get_vocabulary()),
        output_dim=128,
        mask_zero=True,
        input_length=max_comment_len,
    ),
    SpatialDropout1D(0.5),
    LSTM(40, return_sequences=True),
    LSTM(40),
    Dense(6, activation='sigmoid'),
])

Training of neural network this is a process which reduce difference between correct result and result of neural network. NN reduce error by gradient descent.

Loss function - a mesure for difference between real and prediction values. There are different loss functions:
binary crossentropy,
categorical crossentropy,
mean squared error.

Gradient descent have different optimizations:
SGD,
Adam,
Momentum,
AdaGrad.

model_lstm.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy', 'AUC'])

ModelCheckpoint callback is used in conjunction with training using model.fit() to save a model or weights (in a checkpoint file) at some interval, so the model or weights can be loaded later to continue the training from the state saved.

model_lstm_save_path = '/kaggle/working/toxic_model_lstm'
checkpoint_callback_lstm = ModelCheckpoint(
    model_lstm_save_path,
    monitor='val_accuracy',
    save_best_only=True,
    verbose=1,
    save_format='tf',
)

Get data for training

x_train = train_data['comment_text'].values
y_train = train_data[['toxic', 'severe_toxic', 'obscene', 'threat', 'insult', 'identity_hate']].values

Fit of the model

history_lstm = model_lstm.fit(
    x_train,
    y_train,
    epochs=5, # 15
    batch_size=512,
    validation_split=0.2,
    callbacks=[checkpoint_callback_lstm],
)

Result of training

plt.plot(history_lstm.history['accuracy'],
         label='Training accuracy')
plt.plot(history_lstm.history['val_accuracy'],
         label='Validation accuracy')
plt.xlabel('Epochs')
plt.ylabel('Accuracy')
plt.legend()
plt.show()

Get data for testing

x_test = test_data['comment_text'].values
y_test = test_data[['toxic', 'severe_toxic', 'obscene', 'threat', 'insult', 'identity_hate']].values

Check result

model_lstm.evaluate(x_test, y_test, verbose=1)

Check result of saved model

saved_model = load_model(model_lstm_save_path)
saved_model.evaluate(x_test, y_test, verbose=1)

Compress model

!zip -r /kaggle/working/toxic_model_lstm.zip /kaggle/working/toxic_model_lstm

Check model with production data

quotes = [
    "It takes a great deal of bravery to stand up to your enemies, but a great deal more to stand up to your friends.",
    "No, no. This kind of mark cannot be seen. It lives in your very skin… Love, Harry. Love.",
    "I would trust Hagrid with my own life.",
    "Also, our caretaker, Mr. Filch, has asked me to remind you that the third-floor corridor on the right-hand side is out of bounds to everyone who does not wish to die a most painful death.",
    "It is not our abilities that show what we truly are. It is our choices.",
]
not_toxic_prediction = saved_model.predict(quotes)
for labels in not_toxic_prediction:
    print([ round(lbl, 2) for lbl in labels])

quotes = [
    "Maybe If The Fat Lump Had Given This A Squeeze, He'd Have Remembered To Fall On His Fat Ass.",
    "No One Asked Your Opinion, You Filthy Little Mudblood.",
    "Didn't Mummy Ever Tell You It Was Rude To Eavesdrop, Potter? Oh, Yeah. She Was Dead Before You Could Wipe The Drool Off Your Chin.",
    "You're Gonna Regret This! You And Your Bloody Chicken!",
    "I'm going to kill you, Harry Potter. I'm going to destroy you.",
]
toxic_prediction = saved_model.predict(quotes)
for labels in toxic_prediction:
    print([ round(lbl, 2) for lbl in labels])

Additionally:

How to install Kafka from docker:

Download docker-compose from github - https://github.com/tchiotludo/akhq.
Launch docker-compose:

docker-compose pull
docker-compose up -d

Settings:

KAFKA_LISTENERS: INTERNAL://0.0.0.0:9092,OUTSIDE://0.0.0.0:9094
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,OUTSIDE://localhost:9094
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,OUTSIDE:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL

Sources:

ALEXANDER BURMISTROV - About my 0.9872 single model

Microservices for beginners. Spam service. Python. Scikit-learn. Kafka.

Igor Dubinin — Sun, 26 Nov 2023 12:54:49 +0000

Whole series:

Microservices for beginners

Microservices for beginners. Front-end service. Vue js. Socket.io.

Microservices for beginners. Api Gateway service. Nest js. Kafka.

Microservices for beginners. User service. Nest js. Mongodb. Kafka.

Microservices for beginners. Message service. Nest js. Mongodb. Kafka.

Microservices for beginners. Spam service. Python. Scikit-learn. Kafka.

Microservices for beginners. Toxic service. Python. Tensorflow. Kafka.

Spam service provides spam analysis of messages. I use a publish-subscribe pattern through Kafka message broker. And Scikit-learn model for text analysis.

Full code - link

Whole scheme:

Short description:

User opens the front-end application in the browser and writes messages.
Front-end service emits messages to the api gateway through socket.io.
Api gateway emits messages in the Kafka topic for new messages.
Message service subscribes to the topic with new messages, saves them and publishes events into the topic for saved messages.
Users in the front-end service receive messages in the browser from the Api gateway.
Spam service subscribes to the Kafka topic with saved messages. Scikit-learn model analyzes messages. Messages which were predicted as spam do publish into the Kafka topic for analyzed messages.
Message and User services listen to events with spam and toxic messages, mark them in the database, and apply domain logic after receiving these events.

Scheme of spam service:

from infrastructure.config import getConfig
from spam_module.spam_service import SpamService
from infrastructure.kafka import getKafkaConsumer, getKafkaProducer


def main():
    config = getConfig()
    consumer = getKafkaConsumer(config)
    producer = getKafkaProducer(config)
    spam_service = SpamService()

    print('spam-service started')

    for message in consumer:
        try:
            if not message.value.get('message'):
                continue

            is_spam = spam_service.is_spam(message.value.get('message'))
            if is_spam:
                response = {
                    'id': message.value.get('id'),
                    'user_id': message.value.get('user_id'),
                    'analysis': {
                        'spam': True
                    }
                }
                producer.send(config['KAFKA_ANALYSIS_MESSAGE_TOPIC'], response)

        except Exception as error:
            print('error', error)


if __name__ == '__main__':
    main()

src/infrastructure/config.py - this is the infrastructure helper, provides configuration for service.

import os
from dotenv import dotenv_values


def getConfig():
    config = {
        **dotenv_values('./config/.env.dev'),  # load shared development variables
        **dotenv_values('./config/.env.prod'),  # load sensitive variables
        **os.environ,  # override loaded values with environment variables
    }
    return {
        'KAFKA_URI': config['KAFKA_URI'],
        'KAFKA_READY_MESSAGE_TOPIC': config.get('KAFKA_READY_MESSAGE_TOPIC'),
        'KAFKA_READY_MESSAGE_GROUP': config.get('KAFKA_READY_MESSAGE_GROUP'),
        'KAFKA_ANALYSIS_MESSAGE_TOPIC': config.get('KAFKA_ANALYSIS_MESSAGE_TOPIC'),
        'KAFKA_ANALYSIS_MESSAGE_GROUP': config.get('KAFKA_ANALYSIS_MESSAGE_GROUP'),
    }

src/infrastructure/kafka.py -this is the infrastructure helper, provides connection to Kafka.

import json
from kafka import KafkaConsumer
from kafka import KafkaProducer


def getKafkaConsumer(config):
    return KafkaConsumer(
        config['KAFKA_READY_MESSAGE_TOPIC'],
        group_id=config['KAFKA_READY_MESSAGE_GROUP'],
        bootstrap_servers=[config['KAFKA_URI']],
        value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    )

def getKafkaProducer(config):
    return KafkaProducer(
        bootstrap_servers=[config['KAFKA_URI']],
        value_serializer=lambda v: json.dumps(v).encode('utf-8')
    )

src/spam_module/spam_service.py - this file provides application logic. iHere located functionality for message preparation and prediction by the model - spam_module/sms-spam-bayes.joblib.

import re
import string
from joblib import load


class SpamService:
    empty_message = 'Empty message!'
    model_path = './src/spam_module/sms-spam-bayes.joblib'

    def __init__(self):
        self.model = load(self.model_path)

    def clean_text(self, text):
        text = str(text).lower()
        text = re.sub('\[.*?\]', '', text)
        text = re.sub('https?://\S+|www\.\S+', '', text)
        text = re.sub('<.*?>+', '', text)
        text = re.sub('[%s]' % re.escape(string.punctuation), '', text)
        text = re.sub('\n', '', text)
        text = re.sub('\w*\d\w*', '', text)
        return text

    def is_spam(self, message):
        if not message:
            raise ValueError(self.empty_message)

        message = self.clean_text(message)
        prediction = self.model.predict([message])

        return True if prediction[0] == 1 else False

Scikit-learn model

I used "SMS Spam Collection Dataset" to make simple spam classification. Excellent explanation for text classification for this dataset located here.

How to make a spam detection model? My notebook here.

Import modules

import numpy as np
import pandas as pd
import re
import string

from joblib import dump, load
from sklearn.pipeline import Pipeline
from sklearn.metrics import accuracy_score
from sklearn.naive_bayes import MultinomialNB
from sklearn.metrics import classification_report
from sklearn.model_selection import train_test_split
from sklearn.feature_extraction.text import CountVectorizer
from sklearn.feature_extraction.text import TfidfTransformer

Read data from file

data_csv = pd.read_csv('/kaggle/input/sms-spam-collection-dataset/spam.csv', encoding='latin-1')
data_csv.head()

Create dataframe for analysis

data_csv = data_csv[['v1', 'v2']]
data_csv.shape

Check empty fields

data_na = data_csv['v1'].isna().sum()
data_na

data_na = data_csv['v2'].isna().sum()
data_na

Check duplicates

data_duplicates = data_csv.duplicated(subset=['v2']).sum()
data_duplicates

Remove duplicates

data_csv = data_csv.drop_duplicates(subset=['v2'])
data_csv.shape

Data analysis work with numbers. Add new column - "class", ham will 0, spam will 1

data_csv['class'] = data_csv.apply(lambda row: 1 if row['v1'] == 'spam' else 0, axis=1)
data_csv.head()

Data frame usually not prepared for analysis, and contain a lot of not useful symbols.

data_csv[data_csv['v2'].str.find('http:') > -1].head()

Method for removing useless symbols, from here.

def clean_text(text):
    text = str(text).lower()
    text = re.sub('\[.*?\]', '', text)
    text = re.sub('https?://\S+|www\.\S+', '', text)
    text = re.sub('<.*?>+', '', text)
    text = re.sub('[%s]' % re.escape(string.punctuation), '', text)
    text = re.sub('\n', '', text)
    text = re.sub('\w*\d\w*', '', text)
    return text

Add new column with clean text - "clean_text".

data_csv['clean_text'] = data_csv.apply(lambda row: clean_text(row['v2']), axis=1)
data_csv.head()

Check http links

data_csv[data_csv['clean_text'].str.find('http:') > -1]

Prepare data for analysis. We need data to train model. And data to check working of our model.

train_test_split - Split arrays or matrices into random train and test subsets.

x_train, y_train - data to train model

x_test, y_test - data to check result of model

data_x = data_csv['clean_text']
data_y = data_csv['class']
x_train, x_test, y_train, y_test = train_test_split(data_x, data_y, train_size=0.8, test_size=0.2, random_state=10)

Prediction
CountVectorizer - convert a collection of text documents to a matrix of token counts. Ml algorithms working with numbers, so we need to convert our text to list of numbers.

MultinomialNB - naive bayes classifier for multinomial models. This is old classification algorithm, but working still good for classification issues.

bays_model = Pipeline([
    ('count', CountVectorizer(stop_words='english')),
    ('model', MultinomialNB())
])
bays_model.fit(x_train, y_train)
bayes_y_pred = bays_model.predict(x_test)

Result of classification:

Precision is defined as the number of true positives over the number of true positives plus the number of false positives.

Recall is defined as the number of true positives over the number of true positives plus the number of false negatives.

print(classification_report(y_test, bayes_y_pred, target_names=('spam', 'ham')))

Visualisation of results

import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.metrics import confusion_matrix

f, ax = plt.subplots(figsize=(9, 6))
sns.heatmap(confusion_matrix(y_test, bayes_y_pred), annot=True, fmt="d")

Additionally:

How to install Kafka from docker:

Download docker-compose from github - https://github.com/tchiotludo/akhq.
Launch docker-compose:

docker-compose pull
docker-compose up -d

Settings:

KAFKA_LISTENERS: INTERNAL://0.0.0.0:9092,OUTSIDE://0.0.0.0:9094
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,OUTSIDE://localhost:9094
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,OUTSIDE:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL

Sources:

Natural Language Processing (NLP) for Beginners

Microservices for beginners. Message service. Nest js. Mongodb. Kafka.

Igor Dubinin — Sun, 26 Nov 2023 12:54:25 +0000

Whole series:

Microservices for beginners

Microservices for beginners. Front-end service. Vue js. Socket.io.

Microservices for beginners. Api Gateway service. Nest js. Kafka.

Microservices for beginners. User service. Nest js. Mongodb. Kafka.

Microservices for beginners. Message service. Nest js. Mongodb. Kafka.

Microservices for beginners. Spam service. Python. Scikit-learn. Kafka.

Microservices for beginners. Toxic service. Python. Tensorflow. Kafka.

Microservices for beginners. Common code. Typescript.

Message service provides functionality for messages in application, this service contains all messages and chat rooms. I use Nest.js for backend, Mongo database and Kafka as message broker.

Full code - link

Whole scheme:

Short description:

User opens front-end application in web browser and joins chat room, front-end emits an event to the api gateway by socket.io.
Api gateway gets chat data from the message service by http request and emits this to the front-end.
For the messaging, front end service communicates with api gateway by socket.io.
Api gateway implements publish-subscribe pattern to emit raw message events for listeners, through kafka message broker.
Message service receives raw message events, saves messages and emits events with saved messages.
Api gateway receives saved messages and emits this to the front-end application.
Also message service subscribes to analysis events from spam and toxic services, and saves analises for the messages.

Scheme of message service:

main.ts - initialization of service. I use swagger as documentation and REST client for testing of http requests.



import { AppModule } from './app.module';
import { NestFactory } from '@nestjs/core';
import { ConfigService } from '@nestjs/config';
import * as basicAuth from 'express-basic-auth';
import { SwaggerModule, DocumentBuilder } from '@nestjs/swagger';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  const configService: ConfigService = app.get(ConfigService);

  app.use(
    ['/swagger'],
    basicAuth({
      challenge: true,
      users: {
        [configService.get<string>('DOC_USER')]:
          configService.get<string>('DOC_PASS'),
      },
    }),
  );

  const swaggerConfig = new DocumentBuilder()
    .setTitle('Message Service')
    .setDescription('Message Service API description')
    .setVersion('0.0.1')
    .build();

  const document = SwaggerModule.createDocument(app, swaggerConfig);

  SwaggerModule.setup('swagger', app, document, {
    swaggerOptions: {
      persistAuthorization: true,
    },
  });

  await app.listen(configService.get<string>('APP_PORT'));
}

bootstrap();

app.module.ts - describes modules in service:

ConfigModule - provides configuration functionality
MongooseModule - provides Mongoose ODM functionality
MessagesRepoModule - provides communication with database through Mongoose ODM
MessagesModule - implements application and domain logics of service



import { Module } from '@nestjs/common';
import { MongooseModule } from '@nestjs/mongoose';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { AppService } from './app.service';
import { AppController } from './app.controller';
import { MessagesModule } from './modules/messages/messages.module';
import { MessagesRepoModule } from './modules/messages-repo/messages.repo.module';

@Module({
  imports: [
    ConfigModule.forRoot({
      envFilePath: [
        __dirname + '/../config/.env.prod',
        __dirname + '/../config/.env.dev',
      ],
      isGlobal: true,
    }),
    MongooseModule.forRootAsync({
      imports: [ConfigModule],
      inject: [ConfigService],
      useFactory: async (configService: ConfigService) => {
        return { uri: configService.get<string>('MONGO_URI') };
      },
    }),
    MessagesRepoModule,
    MessagesModule,
  ],
  controllers: [AppController],
  providers: [AppService],
})

export class AppModule {}

Messages module

messages.module.ts - describes messages module



import { Module } from '@nestjs/common';
import { MessagesService } from './messages.service';
import { MessagesController } from './messages.controller';
import { KafkaController } from './kafka.controller';
import { MessagesRepoModule } from 'src/modules/messages-repo/messages.repo.module';

@Module({
  providers: [MessagesService],
  controllers: [MessagesController, KafkaController],
  imports: [MessagesRepoModule],
  exports: [MessagesService],
})
export class MessagesModule {}

messages.controller.ts - provides http requests to receive data about chat rooms



import { Controller, Get, UsePipes, HttpStatus, Query } from '@nestjs/common';
import { ApiOperation, ApiResponse, ApiTags } from '@nestjs/swagger';
import { PrivateRoomQueryDto, RoomDataDto } from 'dto-common';
import { JoiValidationPipe } from 'src/pipes/joi.validation.pipe';
import { MessagesService } from './messages.service';
import { privateRoomQueryJoi } from './messages.joi';

@Controller('messages')
export class MessagesController {
  constructor(private messagesService: MessagesService) {}

  @Get('/get-private-room')
  @ApiOperation({ summary: 'Get user by id' })
  @ApiTags('Users')
  @ApiResponse({
    status: HttpStatus.OK,
    description: 'Success',
    type: RoomDataDto,
  })
  @UsePipes(new JoiValidationPipe(privateRoomQueryJoi))
  getPrivateRoom(@Query() params: PrivateRoomQueryDto): Promise<RoomDataDto> {
    return this.messagesService.getPrivateRoom(params);
  }
}

messages.joi.ts - provides validation for messages.controller api



import * as Joi from 'joi';

export const privateRoomQueryJoi = Joi.object({
  userIds: Joi.array().length(2).items(Joi.string().length(24)),
});

kafka.controller.ts - provides publish-subscribe functionality for communication with other services



import { Kafka, Producer, Consumer, KafkaMessage } from 'kafkajs';
import { ConfigService } from '@nestjs/config';
import {
  Controller,
  OnModuleDestroy,
  OnModuleInit,
  Logger,
} from '@nestjs/common';
import { MessagesService } from './messages.service';
import { MessageWebDto, messageAnalysisDto } from 'dto-common';

@Controller()
export class KafkaController implements OnModuleInit, OnModuleDestroy {
  constructor(
    private configService: ConfigService,
    private messagesService: MessagesService,
  ) {}

  private readonly logger = new Logger(KafkaController.name);

  private readonly kafka: Kafka = new Kafka({
    clientId: 'messages',
    brokers: [this.configService.get<string>('KAFKA_URI')],
  });

  private readonly producer: Producer = this.kafka.producer();
  private readonly consumer: Consumer = this.kafka.consumer({
    groupId: this.configService.get<string>('KAFKA_RAW_MESSAGE_GROUP'),
  });
  private readonly analysisConsumer: Consumer = this.kafka.consumer({
    groupId: this.configService.get<string>('KAFKA_ANALYSIS_MESSAGE_GROUP'),
  });

  async onModuleInit() {
    try {
      await this.producer.connect();

      await this.consumer.connect();
      await this.consumer.subscribe({
        topic: this.configService.get<string>('KAFKA_RAW_MESSAGE_TOPIC'),
        fromBeginning: true,
      });
      await this.analysisConsumer.subscribe({
        topic: this.configService.get<string>('KAFKA_ANALYSIS_MESSAGE_TOPIC'),
        fromBeginning: true,
      });

      await this.consumer.run({
        eachMessage: async ({ topic, partition, message }) => {
          this.receiveMessage(message);
        },
      });
      await this.analysisConsumer.run({
        eachMessage: async ({ topic, partition, message }) => {
          this.receiveAnalysis(message);
        },
      });
    } catch (error) {
      this.logger.error(error);
    }
  }

  async onModuleDestroy() {
    try {
      await this.producer.disconnect();
      await this.consumer.disconnect();
      await this.analysisConsumer.disconnect();
    } catch (error) {
      this.logger.error(error);
    }
  }

  async receiveMessage(params: KafkaMessage) {
    try {
      const messageValue = JSON.parse(params.value.toString());
      const { uuid, message, room_id, user_id, created_at } = messageValue;
      const readyMessage: MessageWebDto =
        await this.messagesService.receiveMessage({
          uuid,
          message,
          room_id,
          user_id,
          created_at,
        });

      await this.producer.send({
        topic: this.configService.get<string>('KAFKA_READY_MESSAGE_TOPIC'),
        messages: [
          {
            key: room_id,
            value: JSON.stringify(readyMessage),
          },
        ],
      });
    } catch (error) {
      this.logger.error(error);
    }
  }

  async receiveAnalysis(params: KafkaMessage) {
    try {
      const messageValue: { id: string; analysis: messageAnalysisDto } =
        JSON.parse(params.value.toString());
      const { id, analysis } = messageValue;
      await this.messagesService.receiveAnalysis({ id, analysis });
    } catch (error) {
      this.logger.error(error);
    }
  }
}

messages.service.ts - provides application and domain logic



import { Injectable, ForbiddenException } from '@nestjs/common';
import {
  RoomWebDto,
  PrivateRoomQueryDto,
  MessageWebDto,
  RoomDataDto,
  messageAnalysisDto,
} from 'dto-common';
import { MessagesRepoService } from 'src/modules/messages-repo/messages.repo.service';

@Injectable()
export class MessagesService {
  constructor(private messagesRepoService: MessagesRepoService) {}

  async getPrivateRoom(
    param: PrivateRoomQueryDto,
  ): Promise<RoomDataDto | undefined> {
    let privateRoom = await this.messagesRepoService.getPrivateRoom(param);

    if (!privateRoom)
      privateRoom = await this.messagesRepoService.createPrivateRoom(param);

    const messages = await this.messagesRepoService.getRoomMessages({
      roomId: privateRoom.id,
    });

    return {
      room: privateRoom,
      messages: messages?.reverse(),
    };
  }

  async receiveMessage(
    param: MessageWebDto,
  ): Promise<MessageWebDto | undefined> {
    const { id, user_id } = param;
    const messageRoom: RoomWebDto = await this.messagesRepoService.getUserRoom({
      id,
      user_id,
    });

    if (!messageRoom) {
      throw new ForbiddenException(`Room ${id} forbidden for user ${user_id}`);
    }

    const newMessage = await this.messagesRepoService.saveMessage(param);

    return newMessage;
  }

  async receiveAnalysis(param: { id: string; analysis: messageAnalysisDto }) {
    return this.messagesRepoService.saveAnalysis(param);
  }
}

Messages repo module

messages.repo.module.ts - describes module



import { Module } from '@nestjs/common';
import { MongooseModule } from '@nestjs/mongoose';
import { MessagesRepoService } from './messages.repo.service';
import { Message, MessageSchema } from './messages.schema';
import { Room, RoomSchema } from './rooms.schema';

@Module({
  providers: [MessagesRepoService],
  exports: [MessagesRepoService],
  imports: [
    MongooseModule.forFeature([
      { name: Message.name, schema: MessageSchema },
      { name: Room.name, schema: RoomSchema },
    ]),
  ],
})
export class MessagesRepoModule {}

messages.schema.ts - contains Mongoose schema for messages collection in Mongo database.



import { HydratedDocument } from 'mongoose';
import { Prop, Schema, SchemaFactory } from '@nestjs/mongoose';
import { messageAnalysisDto } from 'dto-common';

export type MessageDocument = HydratedDocument<Message>;

@Schema()
export class Message {
  @Prop({ type: String })
  uuid: string;

  @Prop({ type: String })
  message: string;

  @Prop({ type: String })
  room_id: string;

  @Prop({ type: String })
  user_id: string;

  @Prop({ type: Date })
  created_at: Date;

  @Prop({ type: Object })
  analysis?: messageAnalysisDto;
}

export const MessageSchema = SchemaFactory.createForClass(Message);

rooms.schema.ts - contains Mongoose schema for chat rooms collection in Mongo database.



import { HydratedDocument } from 'mongoose';
import { Prop, Schema, SchemaFactory } from '@nestjs/mongoose';

export type RoomDocument = HydratedDocument<Room>;

@Schema()
export class Room {
  @Prop({ type: String })
  type: string;

  @Prop({ type: Array })
  user_ids: string[];

  @Prop({ type: Date })
  created_at: Date;
}

export const RoomSchema = SchemaFactory.createForClass(Room);

messages.repo.service.ts - contains database requests for messages module



import { Model } from 'mongoose';
import { Injectable } from '@nestjs/common';
import { InjectModel } from '@nestjs/mongoose';
import { Room, RoomDocument } from './rooms.schema';
import { Message, MessageDocument } from './messages.schema';
import {
  RoomWebDto,
  MessageWebDto,
  PrivateRoomQueryDto,
  UserRoomQueryDto,
  messageAnalysisDto,
  messageDto,
} from 'dto-common';

@Injectable()
export class MessagesRepoService {
  constructor(
    @InjectModel(Room.name) private roomModel: Model<RoomDocument>,
    @InjectModel(Message.name) private messageModel: Model<MessageDocument>,
  ) {}

  getWebRoomDto(roomDoc: RoomDocument): RoomWebDto {
    const { _id, user_ids, type, created_at } = roomDoc;
    return {
      id: _id.toString(),
      user_ids: user_ids.map((id) => id.toString()),
      type,
      created_at,
    };
  }

  getWebMessageDto(messageDoc: MessageDocument): MessageWebDto {
    const { _id, uuid, message, room_id, user_id, created_at } = messageDoc;
    return {
      id: _id.toString(),
      uuid,
      message,
      room_id: room_id.toString(),
      user_id: user_id.toString(),
      created_at,
    };
  }

  async getPrivateRoom(
    param: PrivateRoomQueryDto,
  ): Promise<RoomWebDto | undefined> {
    const room: RoomDocument = await this.roomModel.findOne({
      user_ids: { $all: param.userIds },
      type: 'private',
    });

    return room ? this.getWebRoomDto(room) : room;
  }

  async createPrivateRoom(
    param: PrivateRoomQueryDto,
  ): Promise<RoomWebDto | undefined> {
    const room: RoomDocument = await this.roomModel.create({
      type: 'private',
      created_at: new Date(),
      user_ids: param.userIds,
    });
    await room.save();

    return this.getWebRoomDto(room);
  }

  async getUserRoom(param: UserRoomQueryDto): Promise<RoomWebDto | undefined> {
    const { id, user_id } = param;
    const room: RoomDocument = await this.roomModel.findOne({
      id,
      user_ids: user_id,
    });

    return room ? this.getWebRoomDto(room) : room;
  }

  async saveMessage(param: MessageWebDto): Promise<MessageWebDto | undefined> {
    const { uuid, message, room_id, user_id, created_at } = param;
    const newMessage: MessageDocument = new this.messageModel({
      uuid,
      message,
      room_id,
      user_id,
      created_at,
    });
    await newMessage.save();

    return this.getWebMessageDto(newMessage);
  }

  async getRoomMessages(param: {
    roomId: string;
  }): Promise<MessageWebDto[] | undefined> {
    const messages: MessageWebDto[] = await this.messageModel
      .find({ room_id: param.roomId })
      .sort({ _id: -1 })
      .limit(100);

    return messages?.length
      ? messages.map((message: MessageDocument) =>
          this.getWebMessageDto(message),
        )
      : [];
  }

  async saveAnalysis(param: { id: string; analysis: messageAnalysisDto }) {
    const { id, analysis } = param;

    const message = await this.messageModel.findById(id);
    if (!message) {
      return;
    }
    message.analysis = {
      ...(message.analysis || {}),
      ...analysis,
    };
    await message.save();
  }
}

Additionally:

How to install Mongo from docker:

Download docker container, launch with required properties, and open console:



docker pull mongo
docker run -d --name micro-mongo -p 27017:27017 -e MONGO_INITDB_ROOT_USERNAME=micro_user -e MONGO_INITDB_ROOT_PASSWORD=micro_pwd mongo
docker exec -it micro-mongo mongosh

Settings for database and text index:



use admin
db.auth('micro_user', 'micro_pwd')

use micro_messages
db.createUser(
  {
    user: 'micro_user',
    pwd: 'micro_pwd',
    roles: [
       { db: 'micro_messages', role: 'dbAdmin', },
       { db: 'micro_messages', role: 'readWrite', }
    ]
  }
);
db.users.createIndex({ login: 'text' });

How to install Kafka from docker:

Download docker-compose from github - https://github.com/tchiotludo/akhq.
Launch docker-compose:



docker-compose pull
docker-compose up -d

Settings:



KAFKA_LISTENERS: INTERNAL://0.0.0.0:9092,OUTSIDE://0.0.0.0:9094
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,OUTSIDE://localhost:9094
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,OUTSIDE:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL

Microservices for beginners. User service. Nest js. Mongodb. Kafka.

Igor Dubinin — Sun, 26 Nov 2023 12:53:50 +0000

Whole series:

Microservices for beginners

Microservices for beginners. Front-end service. Vue js. Socket.io.

Microservices for beginners. Api Gateway service. Nest js. Kafka.

Microservices for beginners. User service. Nest js. Mongodb. Kafka.

Microservices for beginners. Message service. Nest js. Mongodb. Kafka.

Microservices for beginners. Spam service. Python. Scikit-learn. Kafka.

Microservices for beginners. Toxic service. Python. Tensorflow. Kafka.

Microservices for beginners. Common code. Typescript.

User service provides functionality for users in application. This service contains all users data, gives the ability to register new users, login in the application, receive information about users, and manage status of user - is active or not. I use Nest.js for backend, Mongo database and Kafka as message broker.

Full code - link

Whole scheme:

Short description:

User opens the front-end application in the browser, and tries to create a new account and login, after the login user can see a list of other users.
Front-end service requests data from the api gateway.
Api gateway sends http requests to the users service.
Also the users service subscribes to events from spam and toxic services, and after several toxic messages users will be deactivated.

Scheme of user service:

main.ts - initialization of service. I use swagger as documentation and REST client for testing of http requests.

import { AppModule } from './app.module';
import { NestFactory } from '@nestjs/core';
import { ConfigService } from '@nestjs/config';
import * as basicAuth from 'express-basic-auth';
import { SwaggerModule, DocumentBuilder } from '@nestjs/swagger';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  const configService: ConfigService = app.get(ConfigService);

  app.use(
    ['/swagger'],
    basicAuth({
      challenge: true,
      users: {
        [configService.get<string>('DOC_USER')]:
          configService.get<string>('DOC_PASS'),
      },
    }),
  );

  const swaggerConfig = new DocumentBuilder()
    .setTitle('User Service')
    .setDescription('User Service API description')
    .setVersion('0.0.1')
    .build();

  const document = SwaggerModule.createDocument(app, swaggerConfig);

  SwaggerModule.setup('swagger', app, document, {
    swaggerOptions: {
      persistAuthorization: true,
    },
  });

  await app.listen(configService.get<string>('APP_PORT'));
}
bootstrap();

app.module.ts - contains list of modules:

ConfigModule - provides configuration, from config/.env.dev for dev, or environment variables.
MongooseModule - Mongoose ODM for database interactions.
UsersRepoModule - I put all database requests in one separated module.
UsersModule - provides functionality for management of users.

import { Module } from '@nestjs/common';
import { AppController } from './app.controller';
import { AppService } from './app.service';
import { UsersModule } from './modules/users/users.module';
import { UsersRepoModule } from 'src/modules/users-repo/users.repo.module';
import { MongooseModule } from '@nestjs/mongoose';
import { ConfigModule, ConfigService } from '@nestjs/config';

@Module({
  imports: [
    ConfigModule.forRoot({
      envFilePath: [
        __dirname + '/../config/.env.prod',
        __dirname + '/../config/.env.dev',
      ],
      isGlobal: true,
    }),
    MongooseModule.forRootAsync({
      imports: [ConfigModule],
      inject: [ConfigService],
      useFactory: async (configService: ConfigService) => ({
        uri: configService.get<string>('MONGO_URI'),
      }),
    }),
    UsersRepoModule,
    UsersModule,
  ],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}

Users repo module

users.repo.module.ts - describes module.

import { Module } from '@nestjs/common';
import { MongooseModule } from '@nestjs/mongoose';
import { UsersRepoService } from './users.repo.service';
import { UserSchema, User } from './users.repo.schema';

@Module({
  providers: [UsersRepoService],
  exports: [UsersRepoService],
  imports: [
    MongooseModule.forFeature([{ name: User.name, schema: UserSchema }]),
  ],
})
export class UsersRepoModule {}

users.repo.service.ts - contains database operations through Mongoose ODM

import { Model } from 'mongoose';
import { Injectable } from '@nestjs/common';
import { InjectModel } from '@nestjs/mongoose';
import { User, UserDocument } from './users.repo.schema';
import {
  FindAllDto,
  FindByIdsDto,
  userDto,
  repoRegistrationParamDto,
  usersListDto,
} from 'dto-common';
import { saveAnalysisDto, findActiveUserDto } from 'src/dto/index.dto';

@Injectable()
export class UsersRepoService {
  constructor(@InjectModel(User.name) private userModel: Model<UserDocument>) {}

  getUserDto(user: UserDocument | undefined): userDto | undefined {
    if (user) {
      const {
        _id,
        login,
        password,
        salt,
        email,
        active,
        created_at,
        analysis,
      } = user;

      return {
        id: _id.toString(),
        login,
        password,
        salt,
        email,
        active,
        created_at,
        analysis,
      };
    }
  }

  /**
   * Find active user
   */
  async findActiveUser(param: findActiveUserDto): Promise<userDto | undefined> {
    const { id, login } = param;
    if (!id && !login) {
      return undefined;
    }

    const user = await this.userModel.findOne({
      ...(id ? { _id: id } : {}),
      ...(login ? { login } : {}),
      active: true,
    });

    return this.getUserDto(user);
  }

  /**
   * Find user by id
   */
  async findById(param: { id: string }): Promise<userDto | undefined> {
    if (!param.id) {
      return;
    }
    const { id } = param;
    const user = await this.userModel.findOne({ _id: id });

    return this.getUserDto(user);
  }

  /**
   * Find user by login
   */
  async findByLogin(param: { login: string }): Promise<userDto | undefined> {
    if (!param.login) {
      return;
    }
    const { login } = param;
    const user = await this.userModel.findOne({ login });

    return this.getUserDto(user);
  }

  /**
   * Find users for list view
   */
  async findAll(param: FindAllDto): Promise<usersListDto> {
    const {
      skip = 0,
      limit = 2,
      sortAsc = 'asc',
      login = '',
      excludeIds,
    } = param;

    const sortField =
      param.sortField === 'id' || !param.sortField ? '_id' : param.sortField;

    const users = await this.userModel
      .find({
        active: true,
        ...(login ? { $text: { $search: login } } : {}),
        ...(excludeIds ? { _id: { $nin: excludeIds } } : {}),
      })
      .sort({ [sortField]: sortAsc === 'asc' ? 1 : -1 })
      .skip(skip)
      .limit(limit);

    const count = await this.userModel.count({
      active: true,
    });

    return {
      users: users.map((user) => this.getUserDto(user)),
      count,
    };
  }

  /**
   * Create user
   */
  async registration(
    registrationDto: repoRegistrationParamDto,
  ): Promise<userDto | undefined> {
    const user = await this.userModel.create(registrationDto);

    return this.getUserDto(user);
  }

  /**
   * Find few users by ids
   */
  async findByIds(param: FindByIdsDto): Promise<usersListDto> {
    const users = await this.userModel.find({
      active: true,
      _id: { $in: param.ids },
    });

    return {
      users: users.map((user) => this.getUserDto(user)),
      count: users.length,
    };
  }

  /**
   * Save analysis of messages
   */
  async saveAnalysis(param: saveAnalysisDto): Promise<any> {
    const { id, analysis, active } = param;
    await this.userModel.updateOne(
      { _id: id },
      {
        $set: {
          analysis,
          ...(typeof active === 'boolean' ? { active } : {}),
        },
      },
    );
  }
}

users.repo.schema.ts - schema for the users collection.

import { HydratedDocument } from 'mongoose';
import { Prop, Schema, SchemaFactory } from '@nestjs/mongoose';
import { userAnalysisDto } from 'dto-common';

export type UserDocument = HydratedDocument<User>;

@Schema()
export class User {
  @Prop({ type: String, required: true, unique: true })
  login: string;

  @Prop({ type: String, required: true, unique: true })
  email: string;

  @Prop({ type: String, required: true })
  password: string;

  @Prop({ type: String, required: true })
  salt: string;

  @Prop({ type: Boolean })
  active: boolean;

  @Prop({ type: Date })
  created_at: Date;

  @Prop({ type: Object })
  analysis: userAnalysisDto;
}

export const UserSchema = SchemaFactory.createForClass(User);

Users module

users.module.ts - describes the module.

import { Module } from '@nestjs/common';
import { UsersHelper } from './users.helper';
import { UsersService } from './users.service';
import { UsersController } from './users.controller';
import { KafkaController } from './kafka.controller';
import { UsersRepoModule } from 'src/modules/users-repo/users.repo.module';

@Module({
  providers: [UsersService, UsersHelper],
  exports: [UsersService],
  controllers: [UsersController, KafkaController],
  imports: [UsersRepoModule],
})
export class UsersModule {}

users.controller.ts - provides REST API for front-end service

import {
  Controller,
  Get,
  Param,
  UsePipes,
  Post,
  Body,
  HttpStatus,
  BadRequestException,
  Query,
} from '@nestjs/common';
import { ApiOperation, ApiResponse, ApiTags } from '@nestjs/swagger';
import {
  FindAllDto,
  FindByIdDto,
  WebUserDto,
  WebUsersAllDto,
  WebLoginParamDto,
  WebRegistrationParamDto,
  FindByIdsDto,
} from 'dto-common';
import { JoiValidationPipe } from 'src/pipes/joi.validation.pipe';
import { UsersService } from './users.service';
import {
  loginJoi,
  findAllJoi,
  findByIdJoi,
  findByIdsJoi,
  registrationJoi,
} from './users.joi';

@Controller('users')
export class UsersController {
  constructor(private usersService: UsersService) {}

  @Post('/login')
  @ApiTags('Authorization')
  @ApiOperation({ summary: 'Login' })
  @ApiResponse({
    status: HttpStatus.OK,
    description: 'Success',
    type: WebUserDto,
  })
  @UsePipes(new JoiValidationPipe(loginJoi))
  async login(@Body() body: WebLoginParamDto): Promise<WebUserDto> {
    return this.usersService.login(body);
  }

  @Post('/registration')
  @ApiTags('Authorization')
  @ApiOperation({ summary: 'Registration of new user' })
  @ApiResponse({
    status: HttpStatus.OK,
    description: 'Success',
    type: WebUserDto,
  })
  @UsePipes(new JoiValidationPipe(registrationJoi))
  async registration(
    @Body() body: WebRegistrationParamDto,
  ): Promise<WebUserDto> {
    try {
      return await this.usersService.registration(body);
    } catch (error) {
      if (error?.code === 11000) {
        throw new BadRequestException('Duplicate user', {
          cause: error,
          description: 'User with same login or email already exists',
        });
      }

      throw new BadRequestException(error.message, {
        cause: error,
        description: error.message,
      });
    }
  }

  @Get('/find-all')
  @ApiTags('Users')
  @ApiOperation({ summary: 'List of users' })
  @ApiResponse({
    status: HttpStatus.OK,
    description: 'Success',
    type: WebUsersAllDto,
  })
  @UsePipes(new JoiValidationPipe(findAllJoi))
  findAll(@Query() params: FindAllDto): Promise<WebUsersAllDto> {
    if (params.excludeIds && typeof params.excludeIds === 'string') {
      params.excludeIds = [params.excludeIds];
    }
    return this.usersService.findAll(params);
  }

  @Get('/find-one/:id')
  @ApiOperation({ summary: 'Get user by id' })
  @ApiTags('Users')
  @ApiResponse({
    status: HttpStatus.OK,
    description: 'Success',
    type: WebUserDto,
  })
  @UsePipes(new JoiValidationPipe(findByIdJoi))
  findOne(@Param() params: FindByIdDto): Promise<WebUserDto> {
    return this.usersService.findById(params);
  }

  @Get('/find-by-ids')
  @ApiTags('Users')
  @ApiOperation({ summary: 'List of users by ids' })
  @ApiResponse({
    status: HttpStatus.OK,
    description: 'Success',
    type: WebUsersAllDto,
  })
  @UsePipes(new JoiValidationPipe(findByIdsJoi))
  findByIds(@Query() params: FindByIdsDto): Promise<WebUsersAllDto> {
    if (params.ids && typeof params.ids === 'string') {
      params.ids = [params.ids];
    }

    return this.usersService.findByIds(params);
  }
}

users.joi.ts - provides validation for users.controller

import * as Joi from 'joi';

export const registrationJoi = Joi.object({
  login: Joi.string().min(3).max(100).required(),
  email: Joi.string().max(100).required().email(),
  password: Joi.string().max(100).required(),
  passwordRepeat: Joi.ref('password'),
}).with('password', 'passwordRepeat');

export const loginJoi = Joi.object({
  login: Joi.string().max(100).required(),
  password: Joi.string().max(100).required(),
});

export const findByIdJoi = Joi.object({
  id: Joi.string().length(24).required(),
});

export const findAllJoi = Joi.object({
  excludeIds: Joi.alternatives().try(
    Joi.array().items(Joi.string()),
    Joi.string(),
  ),
  login: Joi.string().min(3).max(100),
  skip: Joi.number().min(0),
  limit: Joi.number().min(1).max(100),
  sortField: Joi.string().valid(...['id', 'login']),
  sortAsc: Joi.string().valid(...['asc', 'desc']),
});

export const findByIdsJoi = Joi.object({
  ids: Joi.array().min(1).items(Joi.string().length(24)),
});

kafka.controller.ts - subscribes to the kafka topic and listens to events from toxic and spam services.

import { ConfigService } from '@nestjs/config';
import { Kafka, Producer, Consumer, KafkaMessage } from 'kafkajs';
import {
  Controller,
  OnModuleDestroy,
  OnModuleInit,
  Logger,
} from '@nestjs/common';
import { messageAnalysisDto } from 'dto-common';
import { UsersService } from './users.service';

@Controller()
export class KafkaController implements OnModuleInit, OnModuleDestroy {
  constructor(
    private configService: ConfigService,
    private usersService: UsersService,
  ) {}

  private readonly logger = new Logger(KafkaController.name);

  private readonly kafka: Kafka = new Kafka({
    clientId: 'user_service',
    brokers: [this.configService.get<string>('KAFKA_URI')],
  });

  private readonly analysisConsumer: Consumer = this.kafka.consumer({
    groupId: this.configService.get<string>('KAFKA_ANALYSIS_MESSAGE_GROUP'),
  });

  async onModuleInit() {
    try {
      await this.analysisConsumer.subscribe({
        topic: this.configService.get<string>('KAFKA_ANALYSIS_MESSAGE_TOPIC'),
        fromBeginning: true,
      });
      await this.analysisConsumer.run({
        eachMessage: async ({ topic, partition, message }) => {
          this.receiveAnalysis(message);
        },
      });
    } catch (error) {
      this.logger.error(error);
    }
  }

  async onModuleDestroy() {
    try {
      await this.analysisConsumer.disconnect();
    } catch (error) {
      this.logger.error(error);
    }
  }

  async receiveAnalysis(params: KafkaMessage) {
    try {
      const messageValue: { user_id: string; analysis: messageAnalysisDto } =
        JSON.parse(params.value.toString());
      const { user_id: id, analysis } = messageValue;
      await this.usersService.receiveAnalysis({ id, analysis });
    } catch (error) {
      this.logger.error(error);
    }
  }
}

users.service.ts - contains domain and application logic for users service. For example I want to block a user if he writes more than 5 toxic or spam messages, kafka.controller launches this method for every message from kafka topic, and users.service uses users-repo to manage user data.

import { BadRequestException, Injectable } from '@nestjs/common';
import {
  WebUserDto,
  FindByIdDto,
  WebLoginParamDto,
  WebRegistrationParamDto,
  FindAllDto,
  WebUsersAllDto,
  FindByIdsDto,
  userDto,
  messageAnalysisDto,
  userAnalysisDto,
} from 'dto-common';
import { saveAnalysisDto } from 'src/dto/index.dto';
import { UsersHelper } from './users.helper';
import { ConfigService } from '@nestjs/config';
import { UsersRepoService } from 'src/modules/users-repo/users.repo.service';

@Injectable()
export class UsersService {
  constructor(
    private usersRepoService: UsersRepoService,
    private usersHelper: UsersHelper,
    private configService: ConfigService,
  ) {}

  async login(loginDto: WebLoginParamDto): Promise<WebUserDto> {
    if (!loginDto.login) {
      throw new BadRequestException('User not found');
    }
    const user = await this.usersRepoService.findByLogin(loginDto);
    if (!user) {
      throw new BadRequestException('User not found');
    }
    if (!user.active) {
      throw new BadRequestException('User blocked');
    }

    const isMatch = await this.usersHelper.isMatchPassword({
      passwordTyped: loginDto.password,
      salt: user.salt,
      password: user.password,
    });
    if (!isMatch) {
      throw new BadRequestException('Password invalid');
    }
    const { id, login, email, active, created_at } = user;

    return { id: id, login, email, active, created_at };
  }

  async registration(
    registrationDto: WebRegistrationParamDto,
  ): Promise<WebUserDto> {
    const { password, salt } = await this.usersHelper.generatePassword(
      registrationDto.password,
    );

    const { id, login, email, active, created_at } =
      await this.usersRepoService.registration({
        ...registrationDto,
        password,
        salt,
        active: true,
        created_at: new Date(),
      });
    return { id, login, email, active, created_at };
  }

  async findById(findOneParam: FindByIdDto): Promise<WebUserDto | undefined> {
    const user = await this.usersRepoService.findActiveUser(findOneParam);
    if (!user) {
      return;
    }
    const { id, login, email, active, created_at } = user;
    return { id: id, login, email, active, created_at };
  }

  async findAll(param: FindAllDto): Promise<WebUsersAllDto | undefined> {
    const usersList = await this.usersRepoService.findAll(param);
    usersList.users = usersList.users.map((user: userDto) => {
      const { id, login, email, active, created_at } = user;

      return { id, login, email, active, created_at };
    });

    return usersList;
  }

  async findByIds(param: FindByIdsDto): Promise<WebUsersAllDto | undefined> {
    const usersList = await this.usersRepoService.findByIds(param);
    usersList.users = usersList.users.map((user: userDto) => {
      const { id, login, email, active, created_at } = user;

      return { id, login, email, active, created_at };
    });

    return usersList;
  }

  async receiveAnalysis(param: {
    id: string;
    analysis: messageAnalysisDto;
  }): Promise<any> {
    const { id, analysis } = param;
    const user = await this.usersRepoService.findById({ id });
    if (!user) {
      return;
    }

    const newAnalysis: userAnalysisDto = [
      'spam',
      'toxic',
      'severe_toxic',
      'obscene',
      'threat',
      'insult',
      'identity_hate',
    ].reduce((acc, key) => {
      acc[key] = user.analysis?.[key] || 0;
      if (analysis[key]) {
        acc[key] += 1;
      }
      return acc;
    }, {});

    const violationLimit = this.configService.get<number>('VIOLATION_LIMIT');
    const violationCount = newAnalysis.toxic + newAnalysis.spam;
    const updateParam: saveAnalysisDto = {
      id: user.id,
      analysis: newAnalysis,
    };

    if (violationCount > violationLimit) {
      updateParam.active = false;
    }

    await this.usersRepoService.saveAnalysis(updateParam);
  }
}

users.helper.ts - provides for users.service some application functionality, like hash of password.

import { Injectable } from '@nestjs/common';
import { pbkdf2, randomBytes } from 'node:crypto';
import { promisify } from 'util';
import { generatedPasswordDto, isMatchPasswordDto } from 'src/dto/index.dto';

const pbkdf2Promise = promisify(pbkdf2);

@Injectable()
export class UsersHelper {
  async generatePassword(password: string): Promise<generatedPasswordDto> {
    const salt = randomBytes(16).toString('hex');

    const passwordHashed = await pbkdf2Promise(
      password,
      salt,
      1000,
      32,
      'sha512',
    );

    return {
      password: passwordHashed.toString('hex'),
      salt: salt,
    };
  }

  async isMatchPassword({
    passwordTyped,
    salt,
    password,
  }: isMatchPasswordDto): Promise<boolean> {
    const passwordHashed = await pbkdf2Promise(
      passwordTyped,
      salt,
      1000,
      32,
      'sha512',
    );

    return passwordHashed.toString('hex') === password;
  }
}

Additionally:

How to install Mongo from docker:

Download docker container, launch with required properties, and open console:

docker pull mongo
docker run -d --name micro-mongo -p 27017:27017 -e MONGO_INITDB_ROOT_USERNAME=micro_user -e MONGO_INITDB_ROOT_PASSWORD=micro_pwd mongo
docker exec -it micro-mongo mongosh

Settings for database and text index:

use admin
db.auth('micro_user', 'micro_pwd')
use micro_users
db.createUser(
  {
    user: 'micro_user',
    pwd: 'micro_pwd',
    roles: [
       { db: 'micro_users', role: 'dbAdmin', },
       { db: 'micro_users', role: 'readWrite', }
    ]
  }
);
db.users.createIndex({ login: 'text' });

How to install Kafka from docker:

Download docker-compose from github - https://github.com/tchiotludo/akhq.
Launch docker-compose:

docker-compose pull
docker-compose up -d

Settings:

KAFKA_LISTENERS: INTERNAL://0.0.0.0:9092,OUTSIDE://0.0.0.0:9094
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,OUTSIDE://localhost:9094
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,OUTSIDE:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL

Microservices for beginners. Api Gateway service. Nest js. Kafka.

Igor Dubinin — Sun, 26 Nov 2023 12:53:20 +0000

Whole series:

Microservices for beginners

Microservices for beginners. Front-end service. Vue js. Socket.io.

Microservices for beginners. Api Gateway service. Nest js. Kafka.

Microservices for beginners. User service. Nest js. Mongodb. Kafka.

Microservices for beginners. Message service. Nest js. Mongodb. Kafka.

Microservices for beginners. Spam service. Python. Scikit-learn. Kafka.

Microservices for beginners. Toxic service. Python. Tensorflow. Kafka.

Microservices for beginners. Common code. Typescript.

Api gateway is the pattern of distributed systems which provides a single entry point for backend application. Api gateway provides different features:

Adapter for backend functionality
Authorization and validation
Load balancer
Caching
Logging

Full code - link

Whole scheme:

Short description:

User opens the front-end application in the browser.
User wants to register a new account. Front-end service sends a http request to the api gateway. Api gateway sends a http request to the user service, and provides a response to the front-end service.
User wants to login in the application. Front-end service sends a http request to the api gateway. Api gateway sends a http request with login and password to the user service. After successful response the api gateway generates two JWT tokens, first to verify the user and second for updating the main token every few minutes. After these actions the api gateway gives a response containing two tokens to the front-end service.
User wants to see a list of other users. Front-end service sends a http request to the api gateway. Api gateway requests data from the user service, and provides a response to the front-end service.
User opens the chat room for messaging. Front-end service emits an event to the api gateway through socket.io. Api gateway requests data from message service about a chat room, and data about users from the users service, and provides data to the front-end service.
User sends a message, front-end service emits a message through socket.io. Api gateway emits messages in the Kafka topic for new messages.
Message service subscribes to the topic with new messages, saves them and publishes events into the topic for saved messages.
Users in the front-end service receive messages in the browser from the Api gateway.
Spam service subscribes to the Kafka topic with saved messages. Scikit-learn model analyzes messages. Messages which were predicted as spam do publish into the Kafka topic for analyzed messages.
Toxic service subscribes to the Kafka topic with saved messages. Tensorflow model analyzes messages. Messages which were predicted as toxic do publish into the Kafka topic for analyzed messages.
Message and User services listen to events with spam and toxic messages, mark them in the database, and apply domain logic after receiving these events.

Scheme of the api gateway:

src/main.ts - initialization of the service. I use swagger as documentation and REST client for testing of http requests.

import * as basicAuth from 'express-basic-auth';
import { NestFactory } from '@nestjs/core';
import { ConfigService } from '@nestjs/config';
import { SwaggerModule, DocumentBuilder } from '@nestjs/swagger';
import { AppModule } from './app.module';
import { SocketIoAdapter } from 'src/modules/messages/io.adapter';

async function bootstrap() {
  const app = await NestFactory.create(AppModule, {
    cors: true,
  });
  const configService: ConfigService = app.get(ConfigService);

  app.useWebSocketAdapter(new SocketIoAdapter(app, configService));

  app.use(
    ['/swagger'],
    basicAuth({
      challenge: true,
      users: {
        [configService.get<string>('DOC_USER')]:
          configService.get<string>('DOC_PASS'),
      },
    }),
  );

  const swaggerConfig = new DocumentBuilder()
    .setTitle('Api Gateway')
    .setVersion('0.0.1')
    .addBearerAuth(
      { type: 'http', scheme: 'bearer', bearerFormat: 'JWT' },
      'JWT',
    )
    .build();

  const document = SwaggerModule.createDocument(app, swaggerConfig);

  SwaggerModule.setup('swagger', app, document, {
    swaggerOptions: {
      persistAuthorization: true,
    },
  });

  await app.listen(configService.get<string>('APP_PORT'));
}
bootstrap();

src/app.module.ts - describes modules of service. There are few modules here:

ConfigModule - nest.js module, provides functionality for configuration
AuthModule - provides functionality for registration, authentication and authorization
UsersModule - provides data and functionality for interface of users
MessagesModule - provides data and functionality for interface of chat rooms and messages

import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import { AppController } from './app.controller';
import { AppService } from './app.service';
import { AuthModule } from 'src/modules/auth/auth.module';
import { UsersModule } from 'src/modules/users/users.module';
import { MessagesModule } from 'src/modules/messages/messages.module';

@Module({
  imports: [
    ConfigModule.forRoot({
      envFilePath: [
        __dirname + '/../config/.env.prod',
        __dirname + '/../config/.env.dev',
      ],
      isGlobal: true,
    }),
    AuthModule,
    UsersModule,
    MessagesModule,
  ],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}

Auth module

src/modules/auth/auth.module.ts - describes module.

Imports - list of classes which this module required.

PassportModule - authentication library.
JwtModule - library for JWT token.
HttpModule - axios http library.

Provides - classes which provide data and functionality for the module. These classes are injected into the module, and for testing possible to replace them.

AuthService - provides application logic for the module.
AuthRepo - provides functionality for data management, for the module.
JwtStrategy - provides functionality for verification of the JWT token.
RefreshTokenStrategy - provides functionality for refreshing of the JWT token.

Controllers - list of controllers.

AuthController - provides REST API for the module.

import { JwtModule } from '@nestjs/jwt';
import { Module } from '@nestjs/common';
import { HttpModule } from '@nestjs/axios';
import { PassportModule } from '@nestjs/passport';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { AuthRepo } from './auth.repo';
import { AuthService } from './auth.service';
import { AuthController } from './auth.controller';
import { JwtStrategy } from 'src/infrastructure/jwt/strategy/jwt.strategy';
import { RefreshTokenStrategy } from 'src/infrastructure/jwt/strategy/jwt.refresh-strategy';

@Module({
  imports: [
    PassportModule,
    JwtModule.registerAsync({
      imports: [ConfigModule],
      inject: [ConfigService],
      useFactory: async (configService: ConfigService) => ({
        secret: configService.get<string>('JWT_SECRET'),
        signOptions: {
          expiresIn: configService.get<string>('JWT_EXPIRES'),
        },
      }),
    }),
    HttpModule,
  ],
  providers: [AuthService, AuthRepo, JwtStrategy, RefreshTokenStrategy],
  controllers: [AuthController],
})
export class AuthModule {}

src/modules/auth/auth.controller.ts - provides REST API for registration and authorization.

Methods:

/registration - provides registration of new users in the application.
/login - provides login in the application
/refresh - provides updating of the JWT token.

import {
  Controller,
  Request,
  Post,
  UseGuards,
  UsePipes,
  Body,
  BadRequestException,
  HttpStatus,
} from '@nestjs/common';
import {
  ApiTags,
  ApiOperation,
  ApiResponse,
  ApiBearerAuth,
} from '@nestjs/swagger';
import { AuthService } from './auth.service';
import { JoiValidationPipe } from 'src/infrastructure/pipes/joi.validation.pipe';
import { registrationJoi, loginJoi } from './auth.joi';
import {
  WebLoginParamDto,
  WebUserDto,
  WebRegistrationParamDto,
} from 'dto-common';
import { RefreshTokenGuard } from 'src/infrastructure/jwt/guard/jwt-refresh.guard';
import { WebAccessTokens } from './auth.dto';

@Controller('auth')
export class AuthController {
  constructor(private authService: AuthService) {}

  @Post('/registration')
  @ApiTags('Authorization')
  @ApiOperation({ summary: 'Registration of new user' })
  @ApiResponse({
    status: HttpStatus.OK,
    description: 'Success',
    type: WebUserDto,
  })
  @UsePipes(new JoiValidationPipe(registrationJoi))
  async registration(
    @Body() body: WebRegistrationParamDto,
  ): Promise<WebUserDto> {
    try {
      return await this.authService.registration(body);
    } catch (error) {
      if (error?.code === 11000) {
        throw new BadRequestException('Duplicate user', {
          cause: error,
          description: 'User with same login or email already exists',
        });
      }

      throw new BadRequestException(error.message, {
        cause: error,
        description: error.message,
      });
    }
  }

  @Post('/login')
  @ApiTags('Authorization')
  @ApiOperation({ summary: 'Login in application' })
  @UsePipes(new JoiValidationPipe(loginJoi))
  async login(@Body() body: WebLoginParamDto): Promise<WebAccessTokens> {
    return this.authService.login(body);
  }

  @Post('/refresh')
  @UseGuards(RefreshTokenGuard)
  @ApiTags('Authorization')
  @ApiOperation({ summary: 'Refresh of JWT token' })
  @ApiBearerAuth('JWT')
  async refresh(@Request() req): Promise<WebAccessTokens> {
    return this.authService.refresh(req.user);
  }
}

src/modules/auth/auth.service.ts - provides application logic for the authentication.

Methods:

getTokens - method provides generation JWT tokens
registration - method provides registration new user in the users service
login - method provides a users login in the users service, and returns two JWT tokens
refresh - method requests a users data from users service, after success response returns updated tokens

import { JwtService } from '@nestjs/jwt';
import { ConfigService } from '@nestjs/config';
import { Injectable, UnauthorizedException } from '@nestjs/common';
import {
  WebRegistrationParamDto,
  WebLoginParamDto,
  WebUserDto,
} from 'dto-common';
import { AuthRepo } from './auth.repo';
import { WebAccessTokens } from './auth.dto';

@Injectable()
export class AuthService {
  constructor(
    private configService: ConfigService,
    private jwtService: JwtService,
    private authRepo: AuthRepo,
  ) {}

  // method provides generation JWT tokens
  private async getTokens(user: WebUserDto): Promise<WebAccessTokens> {
    const { id, login, email, active, created_at } = user;
    const [accessToken, refreshToken] = await Promise.all([
      this.jwtService.signAsync(
        { id, login, email, active, created_at },
        {
          secret: this.configService.get<string>('JWT_SECRET'),
          expiresIn: this.configService.get<string>('JWT_EXPIRES'),
        },
      ),
      this.jwtService.signAsync(
        { id, login, email, active, created_at },
        {
          secret: this.configService.get<string>('JWT_REFRESH_SECRET'),
          expiresIn: this.configService.get<string>('JWT_REFRESH_EXPIRES'),
        },
      ),
    ]);

    return {
      accessToken,
      refreshToken,
    };
  }

  // method provides registration new user in the users service
  async registration(
    registrationDto: WebRegistrationParamDto,
  ): Promise<WebUserDto> {
    const { id, login, email, active, created_at } =
      await this.authRepo.registration(registrationDto);
    return { id, login, email, active, created_at };
  }

  // method provides a users login in the users service, and returns two JWT tokens
  async login(loginDto: WebLoginParamDto): Promise<WebAccessTokens> {
    const user = await this.authRepo.login(loginDto);

    const { id, login, email, active, created_at } = user;
    const { accessToken, refreshToken } = await this.getTokens({
      id,
      login,
      email,
      active,
      created_at,
    });

    return { accessToken, refreshToken };
  }

  // method requests a users data from users service, after success response returns updated tokens
  async refresh(user: WebUserDto): Promise<WebAccessTokens> {
    if (!user?.id) {
      throw new UnauthorizedException();
    }

    const activeUser = await this.authRepo.findUserById({
      id: user.id,
    });

    if (!activeUser) {
      throw new UnauthorizedException();
    }

    const { id, login, email, active, created_at } = activeUser;
    const { accessToken, refreshToken } = await this.getTokens({
      id,
      login,
      email,
      active,
      created_at,
    });

    return { accessToken, refreshToken };
  }
}

src/modules/auth/auth.repo.ts - provides data management for the auth service

Methods:

usersRequest - method provides communication with user service through http requests
login - method provides login in the users service and receiving data about user
registration - method provides creation new users in the user service and receiving data about user
findUserById - method provides receiving data about user

import { HttpService } from '@nestjs/axios';
import {
  BadRequestException,
  Injectable,
  InternalServerErrorException,
} from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import {
  FindByIdDto,
  WebLoginParamDto,
  WebRegistrationParamDto,
  WebUserDto,
} from 'dto-common';

@Injectable()
export class AuthRepo {
  constructor(
    private configService: ConfigService,
    private readonly httpService: HttpService,
  ) {}

  errorMessage: string = 'Oops something went wrong';

  // method provides communication with user service through http requests
  private async usersRequest(method: string, url: string, param?: any) {
    try {
      const uri = `${this.configService.get<string>('API_USERS')}${url}`;
      const response = await this.httpService.axiosRef?.[method](uri, param);
      return response.data;
    } catch (error) {
      if (error.response?.data?.statusCode === 400) {
        throw new BadRequestException(error.response.data.message);
      }
      throw new InternalServerErrorException(
        error.response?.data?.message || this.errorMessage,
      );
    }
  }

  // method provides login in the users service and receiving data about user
  async login(param: WebLoginParamDto): Promise<WebUserDto> {
    return await this.usersRequest('post', '/users/login', param);
  }

  // method provides creation new users in the user service and receiving data about user
  async registration(param: WebRegistrationParamDto): Promise<WebUserDto> {
    return await this.usersRequest('post', '/users/registration', param);
  }

  // method provides receiving data about user
  async findUserById(param: FindByIdDto): Promise<WebUserDto> {
    return await this.usersRequest('get', `/users/find-one/${param.id}`);
  }
}

src/modules/auth/auth.joi.ts - provides validation for REST API

import * as Joi from 'joi';

export const registrationJoi = Joi.object({
  login: Joi.string().max(100).required(),
  email: Joi.string().max(100).required().email(),
  password: Joi.string().max(100).required(),
  passwordRepeat: Joi.ref('password'),
}).with('password', 'passwordRepeat');

export const loginJoi = Joi.object({
  login: Joi.string().max(100).required(),
  password: Joi.string().max(100).required(),
});

Users module

src/modules/users/users.module.ts - describes module.

Imports - list of classes which this module required.

HttpModule - axios http library.

Provides - classes which provide data and functionality for the module. These classes are injected into the module, and for testing possible to replace them.

UsersService - provides application logic for the module.
UsersRepo - provides functionality for data management, for the module.

Controllers - list of controllers.

UsersController - provides REST API for the module.

import { Module } from '@nestjs/common';
import { HttpModule } from '@nestjs/axios';
import { UsersService } from 'src/modules/users/users.service';
import { UsersController } from 'src/modules/users/users.controller';
import { UsersRepo } from './users.repo';

@Module({
  imports: [
    HttpModule,
  ],
  providers: [UsersService, UsersRepo],
  controllers: [UsersController],
})
export class UsersModule {}

src/modules/users/users.controller.ts - provides REST API for the module.

Methods:

/find-all - provides the list of users
/find-one/:id - provides detail information about user

import {
  ApiBearerAuth,
  ApiOperation,
  ApiResponse,
  ApiTags,
} from '@nestjs/swagger';
import {
  Controller,
  Get,
  HttpStatus,
  Param,
  Query,
  UseGuards,
  UsePipes,
  Request,
} from '@nestjs/common';
import { JoiValidationPipe } from 'src/infrastructure/pipes/joi.validation.pipe';
import {
  FindAllDto,
  FindByIdDto,
  WebUserDto,
  WebUsersAllDto,
} from 'dto-common';
import { findByIdJoi, findAllJoi } from './users.joi';
import { UsersService } from './users.service';
import { JwtAuthGuard } from 'src/infrastructure/jwt/guard/jwt-auth.guard';

@Controller('users')
export class UsersController {
  constructor(private usersService: UsersService) {}

  @Get('/find-all')
  @ApiTags('Users')
  @ApiOperation({ summary: 'List of users' })
  @ApiBearerAuth('JWT')
  @ApiResponse({
    status: HttpStatus.OK,
    description: 'Success',
    type: WebUsersAllDto,
  })
  @UseGuards(JwtAuthGuard)
  @UsePipes(new JoiValidationPipe(findAllJoi))
  findAll(
    @Query() params: FindAllDto,
    @Request() req,
  ): Promise<WebUsersAllDto> {
    return this.usersService.findAll(params, req.user);
  }

  @Get('/find-one/:id')
  @UseGuards(JwtAuthGuard)
  @UsePipes(new JoiValidationPipe(findByIdJoi))
  @ApiTags('Users')
  @ApiOperation({ summary: 'Get user by id' })
  @ApiBearerAuth('JWT')
  @ApiResponse({
    status: HttpStatus.OK,
    description: 'Success',
    type: WebUserDto,
  })
  findOne(@Param() params: FindByIdDto): Promise<WebUserDto> {
    return this.usersService.findUserById(params);
  }
}

src/modules/users/users.service.ts - provides application logic for the module.

Methods:

findAll - provides the list of users
findUserById - provides detail information about user

import { Injectable } from '@nestjs/common';
import {
  FindAllDto,
  FindByIdDto,
  WebUserDto,
  WebUsersAllDto,
} from 'dto-common';
import { UsersRepo } from './users.repo';

@Injectable()
export class UsersService {
  constructor(private usersRepo: UsersRepo) {}

  // provides the list of users
  async findAll(param: FindAllDto, user: WebUserDto): Promise<WebUsersAllDto> {
    param.excludeIds = [user.id];
    return this.usersRepo.findAll(param);
  }

  // provides detail information about user
  async findUserById(param: FindByIdDto): Promise<WebUserDto> {
    const { id, login, email, active, created_at } =
      await this.usersRepo.findUserById(param);
    return { id, login, email, active, created_at };
  }
}

src/modules/users/users.repo.ts - provides functionality for data management, for the module.

Methods:

usersRequest - provides communication with user service through http requests
findAll - provides receiving users list data from the users service
findUserById - provides receiving user detail data from the users service

import { HttpService } from '@nestjs/axios';
import {
  BadRequestException,
  Injectable,
  InternalServerErrorException,
} from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import {
  FindAllDto,
  FindByIdDto,
  WebUserDto,
  WebUsersAllDto,
} from 'dto-common';

@Injectable()
export class UsersRepo {
  constructor(
    private configService: ConfigService,
    private readonly httpService: HttpService,
  ) {}

  errorMessage: string = 'Oops something went wrong';

  // provides communication with user service through http requests
  async usersRequest(method: string, url: string, param?: any) {
    try {
      const uri = `${this.configService.get<string>('API_USERS')}${url}`;
      const response = await this.httpService.axiosRef?.[method](uri, param);
      return response.data;
    } catch (error) {
      if (error.response?.data?.statusCode === 400) {
        throw new BadRequestException(error.response.data.message);
      }
      throw new InternalServerErrorException(
        error.response?.data?.message || this.errorMessage,
      );
    }
  }

  // provides receiving users list data from the users service
  async findAll(param: FindAllDto): Promise<WebUsersAllDto> {
    const searchParam = new URLSearchParams(
      Object.entries(param).map((p) => p),
    );
    let url = `/users/find-all?${searchParam.toString()}`;
    return await this.usersRequest('get', url);
  }

  // provides receiving user detail data from the users service
  async findUserById(param: FindByIdDto): Promise<WebUserDto> {
    return await this.usersRequest('get', `/users/find-one/${param.id}`);
  }
}

src/modules/users/users.joi.ts - provides validation for the REST API

import * as Joi from 'joi';

export const findByIdJoi = Joi.object({
  id: Joi.string().length(24).required(),
});

export const findAllJoi = Joi.object({
  excludeIds: Joi.alternatives().try(
    Joi.array().items(Joi.string()),
    Joi.string(),
  ),
  login: Joi.string().min(3).max(100),
  skip: Joi.number().min(0),
  limit: Joi.number().min(1).max(100),
  sortField: Joi.string().valid(...['id', 'login']),
  sortAsc: Joi.string().valid(...['asc', 'desc']),
});

Messages module

src/modules/messages/messages.module.ts - describes module.

Imports - list of classes which this module required.

HttpModule - axios http library.

Provides - classes which provide data and functionality for the module. These classes are injected into the module, and for testing possible to replace them.

MessagesGateway - provides functionality for subscription and emitting of the socket.io.
MessagesService - provides application logic for the module.
MessagesRepo - provides functionality for data management, for the module.

import { Module } from '@nestjs/common';
import { MessagesGateway } from './messages.gateway';
import { MessagesService } from './message.service';
import { MessagesRepo } from './message.repo';
import { HttpModule } from '@nestjs/axios';

@Module({
  imports: [HttpModule],
  providers: [MessagesGateway, MessagesService, MessagesRepo],
})
export class MessagesModule {}

src/modules/messages/io.adapter.ts - settings for the socket.io client

import { ServerOptions } from 'socket.io';
import { ConfigService } from '@nestjs/config';
import { IoAdapter } from '@nestjs/platform-socket.io';
import { INestApplicationContext } from '@nestjs/common';

export class SocketIoAdapter extends IoAdapter {
  constructor(
    private app: INestApplicationContext,
    private configService: ConfigService,
  ) {
    super(app);
  }

  createIOServer(port: number, options?: ServerOptions) {
    port = this.configService.get<number>('SOCKET_PORT');
    const server = super.createIOServer(port, {
      cors: { origin: `*` },
      ...options,
    });
    return server;
  }
}

src/modules/messages/messages.gateway.ts - provides functionality for subscription and emitting of the socket.io.

Methods:

onModuleInit - connection to the kafka message broker at the moment of initialization
onModuleDestroy - disconnection from the kafka message broker at the moment of destruction
receiveReadyMessage - method emits the socket.io event after receiving a new message from the kafka
onMessage - listening of the socket.io messages and producing to the kafka
joinPrivateRoom - listening of the join to chat room from the socket.io and emits room information to the front-end service

import { Kafka, Producer, Consumer, KafkaMessage } from 'kafkajs';
import { ConfigService } from '@nestjs/config';
import {
  UseGuards,
  Logger,
  OnModuleInit,
  OnModuleDestroy,
} from '@nestjs/common';
import {
  WebSocketGateway,
  WebSocketServer,
  SubscribeMessage,
  ConnectedSocket,
  MessageBody,
} from '@nestjs/websockets';
import { JwtSocketGuard } from 'src/infrastructure/jwt/guard/jwt.socket.guard';
import { MessagesService } from './message.service';
import { MessageWebDto } from 'dto-common';

@WebSocketGateway()
export class MessagesGateway implements OnModuleInit, OnModuleDestroy {
  constructor(
    private messagesService: MessagesService,
    private configService: ConfigService,
  ) {}

  private readonly logger = new Logger(MessagesGateway.name);

  // initialization of the kafka library
  private readonly kafka: Kafka = new Kafka({
    clientId: 'api-gateway',
    brokers: [this.configService.get<string>('KAFKA_URI')],
  });
  // producer - producing messages in a topic
  private readonly producer: Producer = this.kafka.producer();
  // consumer - consuming messages from a topic
  private readonly consumer: Consumer = this.kafka.consumer({
    groupId: this.configService.get<string>('KAFKA_READY_MESSAGE_GROUP'),
  });

  // web socket
  @WebSocketServer() server;

  // connection to the kafka message broker at the moment of initialization
  async onModuleInit() {
    try {
      // connection
      await this.producer.connect();

      // consumer
      await this.consumer.connect();
      await this.consumer.subscribe({
        topic: this.configService.get<string>('KAFKA_READY_MESSAGE_TOPIC'),
        fromBeginning: true,
      });
      await this.consumer.run({
        eachMessage: async ({ topic, partition, message }) => {
          this.receiveReadyMessage(message);
        },
      });
    } catch (error) {
      this.logger.error(error);
    }
  }

  // disconnection from the kafka message broker at the moment of destruction
  async onModuleDestroy() {
    try {
      await this.producer.disconnect();
      await this.consumer.disconnect();
    } catch (error) {
      this.logger.error(error);
    }
  }

  // method emits the socket.io event after receiving a new message from the kafka
  receiveReadyMessage(kafkaMessage: KafkaMessage) {
    try {
      const messageValue: MessageWebDto = JSON.parse(
        kafkaMessage.value.toString(),
      );
      this.server.to(messageValue.room_id).emit('message', messageValue);
    } catch (error) {
      this.logger.error(error);
    }
  }

  // listening of the socket.io messages and producing to the kafka
  @UseGuards(JwtSocketGuard)
  @SubscribeMessage('message')
  async onMessage(@ConnectedSocket() client: any, @MessageBody() data: any) {
    try {
      const { uuid, message, room_id } = data;
      const { id: userId } = client.handshake.user;
      const rawMessage: MessageWebDto = {
        uuid,
        message,
        room_id,
        user_id: userId,
        created_at: new Date(),
      };

      await this.producer.send({
        topic: this.configService.get<string>('KAFKA_RAW_MESSAGE_TOPIC'),
        messages: [
          {
            key: room_id,
            value: JSON.stringify(rawMessage),
          },
        ],
      });
    } catch (error) {
      this.logger.error(error);
    }
  }

  // listening of the join to chat room from the socket.io and emits room information to the front-end service
  @UseGuards(JwtSocketGuard)
  @SubscribeMessage('joinPrivateRoom')
  async joinPrivateRoom(
    @ConnectedSocket() client: any,
    @MessageBody() data: any,
  ) {
    try {
      const { userId: secondId } = data;
      const { id: currentId } = client.handshake.user;
      const roomData = await this.messagesService.getPrivateRoom({
        userIds: [secondId, currentId],
      });
      const roomUsers = await this.messagesService.getUsersByIds({
        ids: [secondId, currentId],
      });
      const roomId = roomData?.room?.id;

      client.join(roomId);
      client.emit('joinPrivateRoom', {
        users: roomUsers?.users,
        room: roomData?.room,
        messages: roomData?.messages,
      });
      client.to(roomId).emit('joinPrivateRoom', {
        users: roomUsers?.users,
        room: roomData?.room,
        messages: roomData?.messages,
      });
    } catch (error) {
      this.logger.error(error);
    }
  }
}

src/modules/messages/message.service.ts - provides application logic for the module.

Methods:

getPrivateRoom - provides the data about chat room
getUsersByIds - provides the data about users

import { Injectable } from '@nestjs/common';
import {
  FindByIdsDto,
  PrivateRoomQueryDto,
  RoomDataDto,
  WebUsersAllDto,
} from 'dto-common';
import { MessagesRepo } from './message.repo';

@Injectable()
export class MessagesService {
  constructor(private messagesRepo: MessagesRepo) {}

  async getPrivateRoom(param: PrivateRoomQueryDto): Promise<RoomDataDto> {
    return this.messagesRepo.getPrivateRoom(param);
  }
  async getUsersByIds(param: FindByIdsDto): Promise<WebUsersAllDto> {
    return this.messagesRepo.getUsersByIds(param);
  }
}

src/modules/messages/message.repo.ts - provides functionality for data management, for the module.

messagesRequest - provides communication with the messages service through http requests
usersRequest - provides communication with the users service through http requests
getPrivateRoom - receiving a data about chat room from the messages service
getUsersByIds - receiving a data about users from the users service

import { HttpService } from '@nestjs/axios';
import {
  BadRequestException,
  Injectable,
  InternalServerErrorException,
} from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import {
  FindByIdsDto,
  PrivateRoomQueryDto,
  RoomDataDto,
  WebUsersAllDto,
} from 'dto-common';

@Injectable()
export class MessagesRepo {
  constructor(
    private configService: ConfigService,
    private readonly httpService: HttpService,
  ) {}

  errorMessage: string = 'Oops something went wrong';

  // provides communication with the messages service through http requests
  async messagesRequest(method: string, url: string, param?: any) {
    try {
      const uri = `${this.configService.get<string>('API_MESSAGES')}${url}`;
      const response = await this.httpService.axiosRef?.[method](uri, param);
      return response.data;
    } catch (error) {
      if (error.response?.data?.statusCode === 400) {
        throw new BadRequestException(error.response.data.message);
      }
      throw new InternalServerErrorException(
        error.response?.data?.message || this.errorMessage,
      );
    }
  }

  // provides communication with the users service through http requests
  async usersRequest(method: string, url: string, param?: any) {
    try {
      const uri = `${this.configService.get<string>('API_USERS')}${url}`;
      const response = await this.httpService.axiosRef?.[method](uri, param);
      return response.data;
    } catch (error) {
      if (error.response?.data?.statusCode === 400) {
        throw new BadRequestException(error.response.data.message);
      }
      throw new InternalServerErrorException(
        error.response?.data?.message || this.errorMessage,
      );
    }
  }

  // receiving a data about chat room from the messages service
  async getPrivateRoom(param: PrivateRoomQueryDto): Promise<RoomDataDto> {
    const { userIds } = param;
    const searchParam = new URLSearchParams(
      userIds.map((id) => ['userIds', id]),
    );
    let url = `/messages/get-private-room?${searchParam.toString()}`;

    return await this.messagesRequest('get', url);
  }

  // receiving a data about users from the users service
  async getUsersByIds(param: FindByIdsDto): Promise<WebUsersAllDto> {
    const { ids } = param;
    const searchParam = new URLSearchParams(ids.map((id) => ['ids', id]));
    let url = `/users/find-by-ids?${searchParam.toString()}`;

    return await this.usersRequest('get', url);
  }
}

Infrastructure

Joi

src/pipes/joi.validation.pipe.ts - provides functionality for validation.

import { ObjectSchema } from 'joi';
import { PipeTransform, Injectable, BadRequestException } from '@nestjs/common';

@Injectable()
export class JoiValidationPipe implements PipeTransform {
  constructor(private schema: ObjectSchema) {}

  transform(value: any) {
    const { error } = this.schema.validate(value);
    if (error) {
      throw new BadRequestException('Validation failed');
    }
    return value;
  }
}

JWT

src/infrastructure/jwt/strategy/jwt.strategy.ts - strategy for the passport library

import { Injectable } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { ExtractJwt, Strategy } from 'passport-jwt';
import { PassportStrategy } from '@nestjs/passport';

@Injectable()
export class JwtStrategy extends PassportStrategy(Strategy) {
  constructor(private configService: ConfigService) {
    super({
      jwtFromRequest: ExtractJwt.fromAuthHeaderAsBearerToken(),
      ignoreExpiration: false,
      secretOrKey: configService.get<string>('JWT_SECRET'),
    });
  }

  async validate(payload: any) {
    return payload;
  }
}

src/infrastructure/jwt/strategy/jwt.refresh-strategy.ts - strategy for the passport library

import { Request } from 'express';
import { ExtractJwt, Strategy } from 'passport-jwt';
import { Injectable } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { PassportStrategy } from '@nestjs/passport';

@Injectable()
export class RefreshTokenStrategy extends PassportStrategy(
  Strategy,
  'jwt-refresh',
) {
  constructor(private configService: ConfigService) {
    super({
      jwtFromRequest: ExtractJwt.fromAuthHeaderAsBearerToken(),
      secretOrKey: configService.get<string>('JWT_REFRESH_SECRET'),
      passReqToCallback: true,
    });
  }

  validate(req: Request, payload: any) {
    return payload;
  }
}

api-gateway-service/src/infrastructure/jwt/guard/jwt-auth.guard.ts - guard for the passport library

import { Injectable } from '@nestjs/common';
import { AuthGuard } from '@nestjs/passport';

@Injectable()
export class JwtAuthGuard extends AuthGuard('jwt') {}

api-gateway-service/src/infrastructure/jwt/guard/jwt-refresh.guard.ts - guard for the passport library

import { Injectable } from '@nestjs/common';
import { AuthGuard } from '@nestjs/passport';

@Injectable()
export class RefreshTokenGuard extends AuthGuard('jwt-refresh') {}

api-gateway-service/src/infrastructure/jwt/guard/jwt.socket.guard.ts - guard for authorization for a socket.io emits

import * as jwt from 'jsonwebtoken';
import { ConfigService } from '@nestjs/config';
import { WsException } from '@nestjs/websockets';
import { CanActivate, ExecutionContext, Injectable } from '@nestjs/common';

@Injectable()
export class JwtSocketGuard implements CanActivate {
  constructor(private configService: ConfigService) {}

  canActivate(context: ExecutionContext) {
    try {
      const client = context.switchToWs().getClient();
      const data = context.switchToWs().getData();
      const user: any = jwt.verify(
        data.token || '',
        this.configService.get<string>('JWT_SECRET'),
      );
      client.handshake.user = user;
      return true;
    } catch (error) {
      throw new WsException(error.message);
    }
  }
}

Additionally:

How to install Kafka from docker:

Download docker-compose from github - https://github.com/tchiotludo/akhq.
Launch docker-compose:

docker-compose pull
docker-compose up -d

Settings:

KAFKA_LISTENERS: INTERNAL://0.0.0.0:9092,OUTSIDE://0.0.0.0:9094
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,OUTSIDE://localhost:9094
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,OUTSIDE:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL

Microservices for beginners. Front-end service. Vue js. Socket.io.

Igor Dubinin — Sun, 26 Nov 2023 12:52:38 +0000

Whole series:

Microservices for beginners

Microservices for beginners. Front-end service. Vue js. Socket.io.

Microservices for beginners. Api Gateway service. Nest js. Kafka.

Microservices for beginners. User service. Nest js. Mongodb. Kafka.

Microservices for beginners. Message service. Nest js. Mongodb. Kafka.

Microservices for beginners. Spam service. Python. Scikit-learn. Kafka.

Microservices for beginners. Toxic service. Python. Tensorflow. Kafka.

Full code of example - link

Functional requirements for my front-end service:

1) Provide registration

2) Provide login

3) Provide list of users

4) Provide conversation between users

Additionally:

Application does look for bad guys

Application does look for spammers

5) Not allow blocked users to login again

Implementation

Here I will show how to create a little front-end application, with all described functionality. I use the vue framework and vuetifyjs as a ui library. I use axios for http requests, with JWT token for user access and socket.io for messaging. I use pinia to have storage and share data between different views and components.

First file - main.ts initiation of service.

import { createApp } from "vue";
import { createPinia } from "pinia";

import App from "./App.vue";
import router from "./router";

import "@mdi/font/css/materialdesignicons.css";
import "vuetify/styles";
import { createVuetify } from "vuetify";
import * as components from "vuetify/components";
import * as directives from "vuetify/directives";

const vuetify = createVuetify({
  components,
  directives,
});

const app = createApp(App);
app.config.errorHandler = (error, vm, info) => {
  console.log("error", error, "info", info);
};
app.use(createPinia());
app.use(router);
app.use(vuetify);
app.mount("#app");

App.vue - this is the layout. Layout contains two components SideMenu.vue and HeaderAlert.vue.

<script lang="ts">
import SideMenu from "./components/SideMenu.vue";
import HeaderAlert from "./components/HeaderAlert.vue";
import { useHeaderAlertStore } from "./stores/header-alert";

export default {
  data: () => ({
    toggle: true,
  }),
  methods: {
    toggleDrawer() {
      this.toggle = !this.toggle;
    },
  },
  components: { SideMenu, HeaderAlert },
  errorCaptured(error: any) {
    const headerAlertStore = useHeaderAlertStore();
    headerAlertStore.setError(error);
    if (error.response?.status === 401) {
      this.$router.push({ name: "login" });
    }
  },
};
</script>

<template>
  <v-app>
    <!-- header -->
    <v-app-bar>
      <v-app-bar-nav-icon
        variant="text"
        @click.stop="toggleDrawer"
      ></v-app-bar-nav-icon>

      <v-btn icon>
        <v-icon>mdi-magnify</v-icon>
      </v-btn>

      <v-btn icon>
        <v-icon>mdi-heart</v-icon>
      </v-btn>
    </v-app-bar>

    <!-- side menu -->
    <SideMenu v-model="toggle"></SideMenu>

    <!--  content -->
    <v-main>
      <v-container>
        <HeaderAlert></HeaderAlert>
        <RouterView />
      </v-container>
    </v-main>
  </v-app>
</template>

HeaderAlert.vue - shows notifications on the top of the page.

<script lang="ts">
import { useHeaderAlertStore } from "../stores/header-alert";

export default {
  data: function () {
    const headerAlertStore = useHeaderAlertStore();
    return {
      headerAlertStore: headerAlertStore,
    };
  },
};
</script>

<template>
  <v-alert
    v-if="Boolean(headerAlertStore.success)"
    icon="mdi-alert-circle-outline"
    variant="outlined"
    color="success"
    title="Success"
  >
    {{ headerAlertStore.success?.message }}
  </v-alert>

  <v-alert
    v-if="Boolean(headerAlertStore.error)"
    icon="mdi-alert-circle-outline"
    variant="outlined"
    color="error"
    title="Error"
  >
    {{ headerAlertStore.error?.message }}
  </v-alert>

  <br />
</template>

SideMenu.vue - contains links from the left menu of the page, links should be different, depending on the user state - logged or not logged.

<script lang="ts">
import { useAuthStore } from "../stores/auth";

export default {
  data: function () {
    const authStore = useAuthStore();
    return {
      title: "Application",
      authStore,
    };
  },

  methods: {
    logout: function () {
      this.authStore.logout();
      this.$router.push({ name: "login" });
    },
  },
};
</script>

<template>
  <v-navigation-drawer>
    <v-list>
      <v-list-item>
        <v-row no-gutters>
          <v-col cols="3">
            <v-avatar image="/src/assets/logo.svg" rounded="0"></v-avatar>
          </v-col>
          <v-col>
            <v-list-item-title>{{ title }} </v-list-item-title>
            <v-list-item-subtitle>Hello!</v-list-item-subtitle>
          </v-col>
        </v-row>
      </v-list-item>
    </v-list>

    <v-divider></v-divider>

    <v-list>
      <v-list-item title="Home" value="Home" prepend-icon="mdi-home" to="/">
      </v-list-item>

      <v-list-item
        v-if="!authStore?.auth"
        title="Login"
        value="Login"
        prepend-icon="mdi-login"
        to="/login"
      >
      </v-list-item>

      <v-list-item
        v-if="!authStore?.auth"
        title="Registration"
        value="Registration"
        prepend-icon="mdi-account-plus"
        to="/registration"
      >
      </v-list-item>

      <v-list-item
        v-if="authStore?.auth"
        title="Users"
        value="Users"
        prepend-icon="mdi-account-multiple"
        to="/users-list"
      >
      </v-list-item>
    </v-list>

    <v-divider></v-divider>

    <v-list v-if="authStore?.auth">
      <v-list-item
        title="Logout"
        value="Logout"
        prepend-icon="mdi-logout"
        @click="logout"
      >
      </v-list-item>
    </v-list>
  </v-navigation-drawer>
</template>

User login happens in the LoginView.vue, but I need to change links in the side menu for logged users. To have actual user's state I will use stores/auth.ts

import { defineStore } from "pinia";

export const useAuthStore = defineStore("counter", {
  state: () => {
    const accessToken: string | null = localStorage.getItem("accessToken");
    const refreshToken: string | null =
      localStorage.getItem("refreshToken");
    return {
      accessToken,
      refreshToken,
      auth: Boolean(accessToken),
    };
  },
  actions: {
    refresh({ accessToken, refreshToken }: { accessToken: string, refreshToken: string }) {
      localStorage.setItem("accessToken", accessToken);
      localStorage.setItem("refreshToken", refreshToken);

      this.accessToken = accessToken;
      this.refreshToken = refreshToken;
    },

    login({ accessToken, refreshToken }: { accessToken: string, refreshToken: string }) {
      localStorage.setItem("accessToken", accessToken);
      localStorage.setItem("refreshToken", refreshToken);

      this.accessToken = accessToken;
      this.refreshToken = refreshToken;
      this.auth = true;
    },

    logout() {
      localStorage.removeItem("accessToken");
      localStorage.removeItem("refreshToken");

      this.accessToken = null;
      this.refreshToken = null;
      this.auth = false;
    },
  },
});

Errors may happen in any view after a user's actions or http request, but I always want to show these at the top of the page. To have data for alerts in the header I will use stores/header-alert.ts.

import { defineStore } from "pinia";

export const useHeaderAlertStore = defineStore("headerAlert", {
  state: () => {
    return {
      error: undefined,
      success: undefined,
    };
  },
  actions: {
    clear(type: string, dt: number) {
      setTimeout(() => {
        if (this[type]?.dt === dt) {
          this[type] = undefined;
        }
      }, 10000);
    },
    setError(error: any) {
      this.success = undefined;

      const dt = Date.now();
      this.error = {
        ...error,
        dt,
        message: error.response?.data?.message || error.message,
      };
      this.clear("error", dt);
    },
    setSuccess(success: any) {
      this.error = undefined;

      const dt = Date.now();
      this.success = {
        dt,
        message: success.response?.data?.message || success.message,
        data: success.response?.data || success.data,
      };

      this.clear("success", dt);
    },
  },
});

Router - router/index.ts contains the list of routes and views for them.

import { createRouter, createWebHistory } from "vue-router";

const router = createRouter({
  history: createWebHistory(import.meta.env.BASE_URL),
  routes: [
    {
      path: "/",
      name: "home",
      component: () => import("../views/HomeView.vue"),
    },
    {
      path: "/login",
      name: "login",
      component: () => import("../views/LoginView.vue"),
    },
    {
      path: "/registration",
      name: "registration",
      component: () => import("../views/RegistrationView.vue"),
    },
    {
      path: "/users-list",
      name: "users-list",
      component: () => import("../views/UsersListView.vue"),
    },
    {
      path: "/chat-private/:userId",
      name: "chat-private",
      component: () => import("../views/ChatPrivateView.vue"),
    },
  ],
});

export default router;

RegistrationView.vue - implements registration of users.

<script lang="ts">
import { registerUser } from "@/repos/users";
import { isEmailValid } from "@/utils/conditions";
import { useHeaderAlertStore } from "@/stores/header-alert";

const headerAlertStore = useHeaderAlertStore();

export default {
  data: () => ({
    valid: false,

    showPwd: false,
    showRPwd: false,

    login: "",
    loginRules: [(value: string) => !!value || "Login is required"],

    email: "",
    emailRules: [
      (value: string) => !!value || "E-mail is required",
      (value: string) => {
        return isEmailValid(value) || "E-mail must be valid";
      },
    ],

    password: "",
    passwordRules: [(value: string) => !!value || "Password is required"],

    passwordRepeat: "",

    terms: false,
    termsRules: [(value: string) => !!value || "Accept of terms is required"],
  }),
  methods: {
    async sendRegistration(event: any) {
      event?.preventDefault();

      try {
        const { valid } = await this.validateField();
        if (!valid) {
          return;
        }
        const { login, email, password, passwordRepeat } = this;
        await registerUser({
          login,
          email,
          password,
          passwordRepeat,
        });

        headerAlertStore.setSuccess({
          message: "Registration completed! Let's try to login!",
        });
        setTimeout(() => {
          this.$router.push({ name: "login" });
        }, 1000);
      } catch (error) {
        headerAlertStore.setError(error);
      }
    },
    passwordRepeatRules(value: string) {
      return value === this.password || "Repeat password please!";
    },
    validateField() {
      return this.$refs.form.validate();
    },
  },
  watch: {
    login: "validateField",
    email: "validateField",
    password: "validateField",
    passwordRepeat: "validateField",
    terms: "validateField",
  },
};
</script>

<template>
  <v-form ref="form" v-model="valid" lazy-validation @submit="sendRegistration">
    <v-card class="mx-auto" title="User Registration">
      <v-container>
        <v-text-field
          v-model="login"
          :rules="loginRules"
          color="primary"
          label="Login"
          variant="underlined"
          required
        ></v-text-field>

        <v-text-field
          v-model="email"
          :rules="emailRules"
          color="primary"
          label="Email"
          variant="underlined"
          required
        ></v-text-field>

        <v-text-field
          v-model="password"
          :rules="passwordRules"
          :append-icon="showPwd ? 'mdi-eye' : 'mdi-eye-off'"
          :type="showPwd ? 'text' : 'password'"
          color="primary"
          label="Password"
          placeholder="Enter your password"
          variant="underlined"
          required
          @click:append="showPwd = !showPwd"
        ></v-text-field>

        <v-text-field
          v-model="passwordRepeat"
          :rules="[passwordRepeatRules]"
          :append-icon="showRPwd ? 'mdi-eye' : 'mdi-eye-off'"
          :type="showRPwd ? 'text' : 'password'"
          color="primary"
          label="Repeat password"
          placeholder="Repeat your password"
          variant="underlined"
          required
          @click:append="showRPwd = !showRPwd"
        ></v-text-field>

        <v-checkbox
          v-model="terms"
          :rules="termsRules"
          color="secondary"
          label="I agree to site terms and conditions"
        ></v-checkbox>
      </v-container>

      <v-divider></v-divider>

      <v-card-actions>
        <v-spacer></v-spacer>

        <v-btn color="primary" type="submit">
          Complete Registration

          <v-icon icon="mdi-chevron-right" end></v-icon>
        </v-btn>
      </v-card-actions>
    </v-card>
  </v-form>
</template>

LoginView.vue - implements login of users.

<script lang="ts">
import { loginUser } from "@/repos/users";
import { useHeaderAlertStore } from "@/stores/header-alert";

const headerAlertStore = useHeaderAlertStore();

export default {
  data: () => ({
    valid: false,

    showPwd: false,

    login: "",
    loginRules: [(value: string) => !!value || "Login is required"],

    password: "",
    passwordRules: [(value: string) => !!value || "Password is required"],
  }),
  methods: {
    async sendLogin(event: any) {
      event.preventDefault();
      const { valid } = await this.validateField();
      if (!valid) {
        return;
      }
      try {
        const { login, password } = this;
        await loginUser({
          login,
          password,
        });

        headerAlertStore.setSuccess({
          message: "Login successful. Welcome!",
        });

        setTimeout(() => {
          this.$router.push({ name: "users-list" });
        }, 1000);
      } catch (error) {
        headerAlertStore.setError(error);
      }
    },
    validateField() {
      return this.$refs.form.validate();
    },
  },
  watch: {
    login: "validateField",
    password: "validateField",
  },
};
</script>

<template>
  <v-form ref="form" v-model="valid" lazy-validation @submit="sendLogin">
    <v-card class="mx-auto" title="Authentication">
      <v-container>
        <v-text-field
          v-model="login"
          :rules="loginRules"
          color="primary"
          label="Login"
          variant="underlined"
          required
        ></v-text-field>

        <v-text-field
          v-model="password"
          :rules="passwordRules"
          :append-icon="showPwd ? 'mdi-eye' : 'mdi-eye-off'"
          :type="showPwd ? 'text' : 'password'"
          color="primary"
          label="Password"
          placeholder="Enter your password"
          variant="underlined"
          required
          @click:append="showPwd = !showPwd"
        ></v-text-field>
      </v-container>

      <v-divider></v-divider>

      <v-card-actions>
        <v-spacer></v-spacer>
        <v-btn color="primary" type="submit">
          Authentication
          <v-icon icon="mdi-chevron-right" end></v-icon>
        </v-btn>
      </v-card-actions>
    </v-card>
  </v-form>
</template>

UsersListView.vue - implements list of users. Here I make the a of users with pagination.

<script lang="ts">
import { getUsersList } from "@/repos/users";
import { useHeaderAlertStore } from "@/stores/header-alert";

const headerAlertStore = useHeaderAlertStore();

export default {
  data: () => ({
    users: [],
    count: 0,
    skip: 0,
    limit: 5,
    page: 1,
    pagesLength: 1,
    error: undefined,
  }),
  methods: {
    async getUsers() {
      try {
        const { users, count } = await getUsersList({
          skip: this.skip,
          limit: this.limit,
        });
        this.users = users;
        this.count = count;
        this.pagesLength = Math.ceil(this.count / this.limit);
      } catch (error: any) {
        headerAlertStore.setError(error);
        if (error.response?.status === 401) {
          this.$router.push({ name: "login" });
        }
      }
    },
  },
  async mounted() {
    this.getUsers();
  },
  watch: {
    page: function (page) {
      this.skip = (page - 1) * this.limit;
      this.getUsers();
    },
  },
};
</script>

<template>
  <v-table hover=true>
    <thead>
      <tr>
        <th class="text-left">Login</th>
        <th class="text-left">Actions</th>
      </tr>
    </thead>
    <tbody>
      <tr v-for="user in users" :key="user.id">
        <td>{{ user.login }}</td>
        <td>
          <v-btn
            prepend-icon="mdi-chat-outline"
            variant="outlined"
            color="primary"
            size="small"
            :to="`chat-private/${user.id}`"
            >Message</v-btn
          >
        </td>
      </tr>
    </tbody>
  </v-table>

  <div class="text-center">
    <v-container>
      <v-row justify="center">
        <v-col cols="8">
          <v-container class="max-width">
            <v-pagination
              class="my-4"
              :length="pagesLength"
              v-model="page"
            ></v-pagination>
          </v-container>
        </v-col>
      </v-row>
    </v-container>
  </div>
</template>

ChatPrivateView.vue - implements messenger for users. I use socket.io for messaging and send the authorization token inside every message, because socket.io has difficulties with updating data in headers.

<script lang="ts">
import { v4 as uuidv4 } from 'uuid';
import { useHeaderAlertStore } from "@/stores/header-alert";
import { io } from "socket.io-client";
import { useAuthStore } from "@/stores/auth";
import { refreshTokenRequest } from '@/utils/axios';
import type { messageDto } from '@/dto/messages';
import type { userDto } from '@/dto/users';

const headerAlertStore = useHeaderAlertStore();
const authStore = useAuthStore();
const apiGatewayUrl = import.meta.env.VITE_SOCKET_URI;

const messages: messageDto[] = [];
const usersById: { [key: string]: userDto } = {};

export default {
  data: () => ({
    socket: io(apiGatewayUrl),
    message: "",
    messages,
    loading: false,
    disabled: false,
    roomId: '',
    uuid: '',
    usersById,
  }),
  async mounted() {
    try {
      this.socket.on("exception", async (error: any) => {
        try {
          if (error.message === 'jwt expired') {
            await refreshTokenRequest();
            if (!this.roomId) {
              this.joinRoom();
            }
            this.sendMessage(undefined);
          } else {
            throw error;
          }
        } catch (err: any) {
          headerAlertStore.setError(err);
          if (err.response?.status === 401) {
            this.$router.push({ name: "login" });
          }
        }
      });

      this.socket.on("connect", () => {
        this.joinRoom();
      });

      this.socket.on("joinPrivateRoom", (data) => {
        this.roomId = data.room.id;
        this.usersById = data.users.reduce((acc: { [key: string]: userDto }, user: userDto) => {
          acc[user.id] = user;
          return acc;
        }, {});

        this.messages = data.messages.map(({ uuid, user_id, message, created_at }: messageDto) => ({
          uuid,
          user_id,
          message,
          created_at,
          login: this.usersById[user_id]?.login,
        }));
        this.scrollDown();
      });

      this.socket.on("message", (data) => {
        const { uuid, user_id, message, created_at, } = data;
        this.messages.push({
          uuid,
          user_id,
          message,
          created_at,
          login: this.usersById[user_id]?.login,
        });

        if (uuid === this.uuid) {
          this.loading = false;
          this.disabled = false;
          this.message = '';
        }

        this.scrollDown();
      });

      this.socket.on("disconnect", () => {
        console.log("Disconnected");
      });


    } catch (error) {
      headerAlertStore.setError(error);
    }
  },
  beforeUnmount() {
    this.socket?.disconnect()
  },
  methods: {
    sendMessage(event: any|undefined) {
      event?.preventDefault();

      if (this.message) {
        this.loading = true;
        this.disabled = true;
        this.uuid = uuidv4();
        this.socket?.emit("message", {
          token: authStore.accessToken,
          message: this.message,
          uuid: this.uuid,
          room_id: this.roomId,
        });
      }

    },
    joinRoom() {
      this.socket.emit("joinPrivateRoom", {
        token: authStore.accessToken,
        userId: this.$route.params.userId,
        type: 'private',
      });
    },
    scrollDown() {
      const el = this.$refs.messages_container;
      if (this.messages?.length > 5) {
        setTimeout(() => {
          el.scrollToIndex(this.messages.length - 1);
        }, 1000);
      }
    }
  },
};
</script>

<template>
  <v-card class="mx-auto" title="Chat">
    <v-card-text>
      <v-virtual-scroll
        :height="300"
        :items="messages"
        ref="messages_container"
      >
        <template v-slot:default="{ item }">
          <v-list-item-title>{{ item.login }}</v-list-item-title>
          <v-list-item-subtitle>{{ new Date(item.created_at).toLocaleString() }}: {{ item.message }}</v-list-item-subtitle>
        </template>
      </v-virtual-scroll>
    </v-card-text>

    <v-card-actions>
      <v-container>
        <v-row dense>
          <v-col cols="12">

            <v-form ref="form" @submit="sendMessage">
              <v-text-field
                ref="message_field"
                :loading="loading"
                :disabled="disabled"
                label="Message"
                append-inner-icon="mdi-send"
                v-model="message"
                @click:append-inner="sendMessage"
              >
              </v-text-field>
            </v-form>

          </v-col>
        </v-row>
      </v-container>
    </v-card-actions>

  </v-card>
</template>

I need to get data for my interface or write something in my application, and I use one file which provides access to data repos/users.ts.

import axios from "axios";
import { authenticatedRequest } from '@/utils/axios';
import { useAuthStore } from "@/stores/auth";
import type { userListDto, usersListQueryDto } from "@/dto/users";

const authStore = useAuthStore();
const apiGatewayUrl = import.meta.env.VITE_API_GATEWAY_URI;


// registration of user
export const registerUser = async (param: any) => {
  const { data } = await axios.post(`${apiGatewayUrl}/auth/registration`, param);
  return data;
};

// login of user
export const loginUser = async (param: any) => {
  const { data } = await axios.post(`${apiGatewayUrl}/auth/login`, param);
  const { accessToken, refreshToken } = data;

  authStore.login({ accessToken, refreshToken });
  return { accessToken, refreshToken };
};

// get list of users
export const getUsersList = async (
  param: usersListQueryDto
): Promise<userListDto> => {
  const url = new URL('/users/find-all', apiGatewayUrl);
  Object.entries(param).forEach(([ key, value ]) => url.searchParams.append(key, value?.toString()));

  return authenticatedRequest("get", url.href);
};

Different repositories may make different http requests with different data, but common part with implementation of http request with JWT token I put into infrastructure file utils/axios.ts

import axios from "axios";
import { useAuthStore } from "@/stores/auth";

const authStore = useAuthStore();
const apiGatewayUrl = import.meta.env.VITE_API_GATEWAY_URI;

// request with Bearer token
const requestWithToken = async (method: string, url: string, param: any) => {
  const config = {
    headers: {
      Authorization: `Bearer ${authStore.accessToken}`,
    },
  };

  const result =
    method === "post"
      ? await axios.post(url, param, config)
      : await axios.get(url, config);
  return result.data;
};

// request for refresh Bearer token
export const refreshTokenRequest = async () => {
  const refreshTokenOld = localStorage.getItem("refreshToken");
  const config = {
    headers: {
      Authorization: `Bearer ${refreshTokenOld}`,
    },
  };

  const result = await axios.post(`${apiGatewayUrl}/auth/refresh`, {}, config);

  const { accessToken, refreshToken } = result.data;
  authStore.refresh({ accessToken, refreshToken });

  return { accessToken, refreshToken };
};

// request with token and refresh old
export const authenticatedRequest = async (
  method: string,
  url: string,
  param?: any
) => {
  try {
    return await requestWithToken(method, url, param);
  } catch (error: any) {
    if (error.response?.status === 401) {
      try {
        await refreshTokenRequest();

        return await requestWithToken(method, url, param);
      } catch (secondError) {
        authStore.logout();
        throw secondError;
      }
    }

    throw error;
  }
};

Microservices for beginners

Igor Dubinin — Sun, 26 Nov 2023 12:51:28 +0000

Whole series:

Microservices for beginners

Microservices for beginners. Front-end service. Vue js. Socket.io.

Microservices for beginners. Api Gateway service. Nest js. Kafka.

Microservices for beginners. User service. Nest js. Mongodb. Kafka.

Microservices for beginners. Message service. Nest js. Mongodb. Kafka.

Microservices for beginners. Spam service. Python. Scikit-learn. Kafka.

Microservices for beginners. Toxic service. Python. Tensorflow. Kafka.

Microservices is the architecture style where an application builds from a set of autonomous services. Another type of application is monolith, monolith contains all functionality in one service. But every service inside of a microservice application is monolith.

Many big applications began from monoliths, and this solution have advantages:

Simple development and debug. We contain all our code in one service, it’s easy to launch and debug in IDE.
Easy testing and monitoring. We can control the state of our application by one health check, all our logs saved in one place. It's easy to test one application without integrations and dependencies from other services.
Data consistency. We use one database and will ensure consistency of our data by using database transactions.

But after few years of working and maintenance, monolith gets some disadvantages:

Technology dependency. We start developing your application in a concrete language and framework, like php yii, so we will support the whole project lifetime of these technologies. And we will be looking for developers with these concrete skills.
Understanding and maintenance. One e-commerce application contains a big codebase, which contains customers, basket, orders, payments, cash receipts, sales, shippings. Tightly coupled code will be difficult for maintaining and understanding for new developers. After development several developers will have conflicts in changes of the same files.
Scaling. After increasing traffic we will increase instances of the same application. We will have one database for application and we should increase the size of our storage and performance of our server.
CI/CD pipelines. We will have several developers or several teams which develop one application. And changes of one feature will block deployment of another feature. Because of the big code base we will have a big count of tests and validations, pipelines will be long and heavy.

Modern fashion architecture style of distributed systems is microservices. It's became popular because of some advantages:

Small size. Because of its small code base it’s easy to understand and develop. One developer may realize his service without knowledge of the whole project.
Agility. Because our application contains small different services, we can deploy new features without blocking other services and teams.
Scalability. After increasing traffic or data, we can increase instances of current service without affecting others.
Reliability. Bug in one service will not affect others, and part of our application will still work.
Different technologies. We can develop new services based on other languages or modern frameworks. Or we can employ developers for new services with knowledge of another language because our services communicate by http or message broker.

But distributed systems contain many disadvantages and challenges:

Complexity. Every service will be easy, but understanding the whole application and coupling services between themselves will be difficult.
Development and testing. Every small service requires data and communication from other services. Testing of applications requires knowledge of whole services and communications between them.
Network. Every service will use others like monolith modules, but will use a network for every method usage.
Data consistency. Every service has his own database. We will need to use several services for our business case and when one of them will not respond we will need to abort others. We will need to apply a difficult approach to consistency and distributed transactions instead of one database transaction.
Different technologies. We can use different technologies, but it will be difficult to have one developer for every technology or highly skilled developers for all of our languages and frameworks.

Web message system

I want to create a message system, which events may happen there?

As a user I want to create my account in the system and login there. In the success case I want to see a success notification, in the error case I want to see an error message.

As a user I want to chat with another user. In the success case I will see our dialog, in the error case, I am not sure what should happen, error message or redirection. Should a system notify a second participant or not?

As a user I want to send a message into chat. In the success case I will see my message. System should send the message to the second participant. As a service owner I want to feel a friendly atmosphere. I want to avoid spam and toxic messages. Message service will send messages into AI services to analyze spam and toxicity.

From these events I'm selecting several services:

Front service - provides user interface in web browser.
User service - provides functionality for users in application.
Message service - provides functionality for conversation.
Spam service - provides spam detection.
Toxic service - provides toxic detection.
API Gateway - provides a single entry point for backend in the application.

Short description:

User opens the front-end application in the browser.
User wants to register a new account. Front-end service sends a http request to the api gateway. Api gateway sends a http request to the user service, and provides a response to the front-end service.
User wants to login in the application. Front-end service sends a http request to the api gateway. Api gateway sends a http request with login and password to the user service. After successful response the api gateway generates two JWT tokens, first to verify the user and second for updating the main token every few minutes. After these actions the api gateway gives a response containing two tokens to the front-end service.
User wants to see a list of other users. Front-end service sends a http request to the api gateway. Api gateway requests data from the user service, and provides a response to the front-end service.
User opens the chat room for messaging. Front-end service emits an event to the api gateway through socket.io. Api gateway requests data from message service about a chat room, and data about users from the users service, and provides data to the front-end service.
User sends a message, front-end service emits a message through socket.io. Api gateway emits messages in the Kafka topic for new messages.
Message service subscribes to the topic with new messages, saves them and publishes events into the topic for saved messages.
Users in the front-end service receive messages in the browser from the Api gateway.
Spam service subscribes to the Kafka topic with saved messages. Scikit-learn model analyzes messages. Messages which were predicted as spam do publish into the Kafka topic for analyzed messages.
Toxic service subscribes to the Kafka topic with saved messages. Tensorflow model analyzes messages. Messages which were predicted as toxic do publish into the Kafka topic for analyzed messages.
Message and User services listen to events with spam and toxic messages, mark them in the database, and apply domain logic after receiving these events.

Sources:

Microsoft azure architecture guide

System-design course

Microservices architecture

Event storming

Event Storming - what it is and why you should use it with Domain-Driven Design

Javascript simple balancing for parallel workers

Igor Dubinin — Thu, 01 Jul 2021 07:31:06 +0000

At current days web services is grown up, mobile application has a thousands of users. Backend is scale up and contain few services that working are parallel. But sometimes parallel instances are not solved a business issues. For example you have business process where is important apply requests one by one. Some item is created, after that, it is updating, and after that, it is finished or cancelled. Your instances could have different performance, or the logic of some methods is more slowly. Because of that the event for cancelling could be launch before the creation of item.

const instanceCount = 3; // count of instances
let instanceFinished = 0; // variable to save count of the finished workers
const storage = {}; // fake storage

// example of one service
class Queue {
    constructor() {
        this.list = []; // list of events in queue
        this.inProcess = false; // status of worker
    }

    // method to add event in queue
    push({ item, worker }) {
        this.list.push({ item, worker });
        this.process();
    }

    // method for launch every event after finish previous
    async process() {
        if (this.inProcess) {
            return;
        }
        if (! this.list.length) {
            instanceFinished += 1;
            if (instanceFinished === instanceCount) {
                console.log('storage', storage);
            }
            return;
        }

        this.inProcess = true;
        const { item, worker } = this.list.shift();
        try {
            await worker(item);
        } catch(e) {
            console.log(e);
        } finally {
            this.inProcess = false;
            this.process();
        }
    }
}

// example of logic with different time of working
class UseCase {
    /**
     * Method to save the item in storage
     * @param {string} key
     */
    async create({ key }) {
        await new Promise((res, rej) => {
            setTimeout(() => {
                if (storage[key]) {
                    return rej(`Item ${key} already exists`);
                }
                storage[key] = { status: 'created' };
                console.log(`Item ${key} created ${Date().toString()}`);
                res();
            }, 300);
        });
    }

    /**
     * Method to update the item
     * @param {string} key
     */
    async update({ key }) {
        await new Promise((res, rej) => {
            setTimeout(() => {
                if (! storage[key]) {
                    return rej(`Item ${key} is not exists`);
                }
                storage[key].status = 'updated';
                console.log(`Item ${key} updated ${Date().toString()}`);
                res();
            }, 200);
        });
    }

    /**
     * Method to cancel the item
     * @param {string} key
     */
    async cancel({ key }) {
        await new Promise((res, rej) => {
            setTimeout(() => {
                if (! storage[key]) {
                    return rej(`Item ${key} is not exists`);
                }
                storage[key].status = 'cancelled';
                console.log(`Item ${key} cancelled ${Date().toString()}`);
                res();
            }, 100);
        });
    }
}

After launching this service you could see that workers try to update or cancel items that was not created because the logic of creation is slower than the logic of updating and cancelling.

const serivce = async () => {
    const useCase = new UseCase();
    const queues = [];

    for (let i = 0; i < instanceCount; i++) {
        queues.push(new Queue());
    }

    for (let i = 1; i < 10; i++) {
        const item = { key: i.toString() };
        queues[0].push({ item, worker: useCase.create });
        queues[1].push({ item, worker: useCase.update });
        queues[2].push({ item, worker: useCase.cancel });
    }
}
serivce();

But if instances will working only with their part of items and every event of item will launch after previous event all be fine.
To separate items by partitions we could use a hash. We should convert key of item into hash number. After take the remaining from division as id of the instance.

/**
 * function for convert string to 32bit integer
 * Source: http://werxltd.com/wp/2010/05/13/javascript-implementation-of-javas-string-hashcode-method/
 * @param {string} str - string for hashing
 */
const hashCode = (str) => {
    var hash = 0;
    if (str.length === 0) {
        return hash;
    }
    for (i = 0; i < str.length; i++) {
        char = str.charCodeAt(i);
        hash = ((hash << 5) - hash) + char;
        hash = hash & hash;
    }
    return hash;
}

const serviceWithHashing = async () => {
    const useCase = new UseCase();
    const queues = [];

    for (let i = 0; i < instanceCount; i++) {
        queues.push(new Queue());
    }

    for (let i = 1; i < 10; i++) {
        const item = { key: i.toString() };
        const queue = queues[hashCode(i.toString()) % instanceCount];
        queue.push({ item, worker: useCase.create });
        queue.push({ item, worker: useCase.update });
        queue.push({ item, worker: useCase.cancel });
    }
}
serviceWithHashing();

This method is not for big distributed systems where count of instances changed dynamically, but could be helpful for launching of few parallel jobs or process.

Javascript red-black tree

Igor Dubinin — Wed, 30 Jun 2021 20:32:50 +0000

Red-back tree is self-balancing binary search tree. Red-black trees are uses at current moment in some computer languages for making key-value structures, like objects or associative arrays.

Properties of red black tree:

1) Node should be red or black and might have two child
2) Root of tree should be black
3) All leaves without value is black
4) Both child of red node is black
5) Every way from parent to child should contain equal number of black nodes

Constants for red-black tree.

const CONSTANTS = {
    RED: 'RED',
    BLACK: 'BLACK',
};

Node for red-black tree.

class Node {
    constructor(param) {
        this.key = param.key || 0;
        this.color = param.color || CONSTANTS.RED;
        this.left = param.left || undefined;
        this.right = param.right || undefined;
        this.parent = param.parent || undefined;
    }
}

Class for red-black tree.

class Tree {
    constructor() {
        this.leaf = new Node({ key: 0, color: CONSTANTS.BLACK });
        this.root = this.leaf;
    }
}

Method to print all nodes from the tree by traverse in deep

printTree() {
    const stack = [
        { node: this.root, str: '' },
    ];

    while (stack.length) {
        // Take last item from stack
        const item = stack.pop();
        // Don't print empty leaf
        if (item.node == this.leaf) {
            continue;
        }
        // Get position of node - left or right
        let position = '';
        if (item.node.parent) {
            position = item.node === item.node.parent.left ? 'L----' : 'R----';
        } else {
            position = 'ROOT-';
        }
        // Print info about node
        console.log(`${item.str}${position} ${item.node.key} (${item.node.color})`);

        // Add node children into stack
        stack.push({ node: item.node.right, str: item.str + '     ' });
        stack.push({ node: item.node.left, str: item.str + ' |   ' });
    }
}

Method to change positions of nodes, the right child will be the new vertex, previous vertex became the left child

/**
* @param {Node} node - vertex for rotation
*/
rotateLeft(node) {
    const vertex = node.right;

    // set new right child for node
    node.right = vertex.left;
    if (vertex.left != this.leaf) {
        vertex.left.parent = node;
    }

    // replace node by new vertex
    vertex.parent = node.parent;
    // if node is root, set new root
    if (! node.parent) {
        this.root = vertex;
    }
    // replace node for parent
    else if (node === node.parent.left) {
        node.parent.left = vertex;
    }
    else {
        node.parent.right = vertex;
    }

    // set left child for vertex - node
    vertex.left = node;
    node.parent = vertex;
}

Method to change positions of nodes, the left child will be the new vertex, previous vertex became the right child

/**
* @param {Node} node - vertex for rotation
*/
rotateRight(node) {
    // left child is new vertex
    const vertex = node.left;

    // node lose left child, we replace it with right child from new vertex
    node.left = vertex.right;
    if (vertex.right != this.leaf) {
        vertex.right.parent = node;
    }

    // new vertex replaces old node
    vertex.parent = node.parent;
    if (! node.parent) {
        this.root = vertex;
    } else if (node == node.parent.right) {
        node.parent.right = vertex;
    } else {
        node.parent.left = vertex;
    }

    // attach right child for new vertex - it is old node
    vertex.right = node;
    node.parent = vertex;
}

Insertion in the binary tree - walk by tree in while cycle, if the key is less that the key of current node, need search in the left node, else in the right node. After insertion we should check balance into the tree.

/**
* @param {number} key - key for new node
*/
insert({ key }) {
    const node = new Node({
        key,
        left: this.leaf,
        right: this.leaf,
    });

    let parent;
    let tmp = this.root;

    // Search of parent for new node
    // we check all nodes while not get an empty leaf
    while (tmp !== this.leaf) {
        parent = tmp;
        // key less that key of current node, we should search in left subtree
        if (node.key < tmp.key) {
            tmp = tmp.left;
        }
        // key bigger that key of current node, we should search in right subtree
        else {
            tmp = tmp.right;
        }
    }

    node.parent = parent;

    // insert node in left or right subtree
    if (! parent) {
        this.root = node;
    } else if (node.key < parent.key) {
        parent.left = node;
    } else {
        parent.right = node;
    }

    // tree has no vertex, node will be root
    if (! node.parent) {
        node.color = CONSTANTS.BLACK;
        return;
    }
    // node has no grandparent, so we have no to balance the tree
    if (! node.parent.parent) {
        return;
    }

    // balancing of tree
    this.balanceInsert(node);
}

Method to balance the tree after inserting.
1) Make the balancing of the tree while the parent of the node is red.
2) If the parent of the node is the left child of his grandparent:
a) if the uncle and the parent are red, we can change colors of the parent and the uncle to black, make the grandparent red, and apply balancing for the grandparent to solve rule 4.
b) if the parent is red and the uncle is black. If the node is the right child, apply balancing to the parent and rotate left.
After make the parent black and the grandparent red. Apply rotate right for the grandparent.
3) If the parent of the node is the right child:
a) If the parent and the uncle is red, we should make them black and make the grandparent red. After will apply balancing to the grandparent.
b) Else, if the node is the left child, we continue balancing from the parent and make left rotation. After set color of the parent as black, make the grandparent red and apply right rotation for the grandparent.
4) Set black color for the root.

/**
* @param {Node} node - node for balancing
*/
balanceInsert(node) {
    // while parent is red
    while (node.parent.color === CONSTANTS.RED) {
        // node parent is left child of grandparent
        if (node.parent === node.parent.parent.left) {
            const uncle = node.parent.parent.right;
            // if uncle and parent are red, need make these black and grandparent red
            if (uncle.color === CONSTANTS.RED) {
                uncle.color = CONSTANTS.BLACK;
                node.parent.color = CONSTANTS.BLACK;
                node.parent.parent.color = CONSTANTS.RED;
                node = node.parent.parent;
            }
            // if parent is red and uncle is black
            else {
                // if node is right child
                if (node === node.parent.right) {
                    node = node.parent;
                    this.rotateLeft(node);
                }
                node.parent.color = CONSTANTS.BLACK;
                node.parent.parent.color = CONSTANTS.RED;
                this.rotateRight(node.parent.parent);
            }
        } else {
            const uncle = node.parent.parent.left;
            if (uncle.color === CONSTANTS.RED) {
                uncle.color = CONSTANTS.BLACK;
                node.parent.color = CONSTANTS.BLACK;
                node.parent.parent.color = CONSTANTS.RED;
                node = node.parent.parent;
            } else {
                if (node == node.parent.left) {
                    node = node.parent;
                    this.rotateRight(node);
                }
                node.parent.color = CONSTANTS.BLACK;
                node.parent.parent.color = CONSTANTS.RED;
                this.rotateLeft(node.parent.parent);
            }
        }

        if (node == this.root) {
            break;
        }
    }

    this.root.color = CONSTANTS.BLACK;
}

Method to find the minimum value for the subtree

/**
* @param {Node} node - node of the tree where we should search the minimum value
*/
minimum(node) {
    while (node.left != this.leaf) {
        node = node.left;
    }
    return node;
}

Method to replace oldNode by newNode

/**
* @param {Node} oldNode - node that should be replaced
* @param {Node} newNode - node that value will be used instead the old node
*/
replace(oldNode, newNode) {
    if (! oldNode.parent) {
        this.root = newNode;
    } else if (oldNode == oldNode.parent.left) {
        oldNode.parent.left = newNode;
    } else {
        oldNode.parent.right = newNode;
    }
    newNode.parent = oldNode.parent;
}

Method to delete the node by key.
If the node that we should remove has only one child, we replace the node by his child. If the node has both children, we should find the minimum child in the right subtree, and use that to replace the node.

/**
* @param {number} key - key for node that should be removed
*/
deleteNode(key) {
    let forRemove = this.leaf;
    let tmp = this.root;

    // searching the node for removing
    while (tmp != this.leaf) {
        if (tmp.key === key) {
            forRemove = tmp;
            break;
        }

        if (tmp.key > key) {
            tmp = tmp.left;
        } else {
            tmp = tmp.right;
        }
    }

    // node is not found
    if (forRemove == this.leaf) {
        console.log('node not found');
        return;
    }

    let minRight = forRemove;
    let minRightColor = minRight.color;
    let newMinRight;

    /*
    if the node for removing has no left child,
    we replace this by its right child
    */
    if (forRemove.left == this.leaf) {
        newMinRight = forRemove.right;
        this.replace(forRemove, forRemove.right);
    }
    /*
    if the node for removing has no right child,
    we replace this by its left child
    */
    else if (forRemove.right == this.leaf) {
        newMinRight = forRemove.left;
        this.replace(forRemove, forRemove.left);
    }
    // if the node for removing have both children
    else {
        minRight = this.minimum(forRemove.right);
        minRightColor = minRight.color;
        newMinRight = minRight.right;

        if (minRight.parent === forRemove) {
            newMinRight.parent = minRight;
        }
        /*
        replace minimum of the right subtree by its right child,
        attach right children from node for removing into the minimum of right subtree
        */
        else {
            this.replace(minRight, minRight.right);
            minRight.right = forRemove.right;
            minRight.right.parent = minRight;
        }

        // attach left children from node for removing into the minimum of right subtree
        this.replace(forRemove, minRight);
        minRight.left = forRemove.left;
        minRight.left.parent = minRight;
        minRight.color = forRemove.color;
    }

    if (minRightColor === CONSTANTS.BLACK) {
        this.balanceDelete(newMinRight);
    }
}

Method to balance the tree after deleting.
1) Make balancing of the tree while the node isn't root of the tree and color of the node is black
2) If the node is the left child of his parent
a) If the brother of the node is red: set color of the brother as black, set color of the parent as red. Apply left rotation to the parent of the node. Set the right child of the parent as the brother.
b) If children of the brother are black: set color of the brother as red and apply balancing to the parent of the node.
c) If color of one child of the brother is red. If color of the right child of the brother is black: set color of the left child as black, set color of the brother as red, apply right rotation to the brother, set the right child of the parent as brother. After, set color of the brother equal to the parent color, set color of the parent as black, set color for the right child of the brother as black. Apply left rotation to the parent of the node. Set the root of the tree as the node.
3) If node is the right child and his brother is the left child.
a) If color of the brother is red. Set color of the brother as black, set color of the parent as red, apply right rotation for the parent of the node, and assign the left child of the parent as brother.
b) If both children of the brother is black. Set color of the brother as red and apply balancing to the parent.
c) If one of child of the brother is red. If the left child of the brother is black, set color of the right child of the brother as black, set color of the brother as red, apply left rotation to the brother, set the left child of the parent as brother. After, set color of the brother as color of the parent. Set color og the parent as black, set color of the left child of the brother as black, Apply right rotation to the parent. Set the root as the node.

/**
* @param {Node} node - node for balancing
*/
balanceDelete(node) {
    while (node != this.root && node.color == CONSTANTS.BLACK) {
        if (node == node.parent.left) {
            let brother = node.parent.right;

            if (brother.color == CONSTANTS.RED) {
                brother.color = CONSTANTS.BLACK;
                node.parent.color = CONSTANTS.RED;
                this.rotateLeft(node.parent);
                brother = node.parent.right;
            }

            if (
                brother.left.color == CONSTANTS.BLACK &&
                brother.right.color == CONSTANTS.BLACK
            ) {
                brother.color = CONSTANTS.RED;
                node = node.parent;
            } else {
                if (brother.right.color == CONSTANTS.BLACK) {
                    brother.left.color = CONSTANTS.BLACK;
                    brother.color = CONSTANTS.RED;
                    this.rotateRight(brother);
                    brother = node.parent.right;
                }

                brother.color = node.parent.color;
                node.parent.color = CONSTANTS.BLACK;
                brother.right.color = CONSTANTS.BLACK;
                this.rotateLeft(node.parent);
                node = this.root;
            }
        } else {
            let brother = node.parent.left
            if (brother.color == CONSTANTS.RED) {
                brother.color = CONSTANTS.BLACK;
                node.parent.color = CONSTANTS.RED;
                this.rotateRight(node.parent);
                brother = node.parent.left;
            }

            if (
                brother.left.color == CONSTANTS.BLACK &&
                brother.right.color == CONSTANTS.BLACK
            ) {
                brother.color = CONSTANTS.RED;
                node = node.parent;
            } else {
                if (brother.left.color == CONSTANTS.BLACK) {
                    brother.right.color = CONSTANTS.BLACK;
                    brother.color = CONSTANTS.RED;
                    this.rotateLeft(brother);
                    brother = node.parent.left;
                }

                brother.color = node.parent.color;
                node.parent.color = CONSTANTS.BLACK;
                brother.left.color = CONSTANTS.BLACK;
                this.rotateRight(node.parent);
                node = this.root;
            }
        }
    }

    node.color = CONSTANTS.BLACK;
}

Example of working for red-black tree

const t = new Tree();

for (let i = 1; i < 20; i++) {
    t.insert({ key: i });
}
t.printTree();

for (let i = 1; i < 20; i++) {
    if (i % 3 === 0) {
        t.deleteNode(i);
    }
}
t.printTree();

JavaScript AVL tree

Igor Dubinin — Fri, 14 May 2021 20:48:37 +0000

Binary search tree

This is tree there every node have only two childs, left and right. Left node contain value less his vertex, right vertex contain value more that parent vertex.

AVL tree

AVL tree is a self-balancing Binary Search Tree where the difference between heights of left and right subtrees cannot be more than one for all nodes. If difference bitween heights is bigger, balancing should fix difference.

Node

For avl tree need to have the balance factor, it is difference between right and left heights.

class Node {
    constructor({ key = 0, height = 1, left, right }) {
        this.key = key;
        this.height = height;
        this.left = left;
        this.right = right;
    }

    getBalanceFactor() {
        const lHeight = this.left ? this.left.height : 0;
        const rHeight = this.right ? this.right.height : 0;
        return rHeight - lHeight;
    }

    fixHeight() {
        const lHeight = this.left ? this.left.height : 0;
        const rHeight = this.right ? this.right.height : 0;
        this.height = (lHeight > rHeight ? lHeight : rHeight) + 1;
    }
}

Balancing

Balancing of tree uses methods of left and right rotation.

Left rotation
Right child became new root
Node lose right child, and left child of old right child became new right child of the node
Old right child lose left child, and node became new left child of the new root

Right rotation
Left child became new root
Node lose left child, and right child of old left child became new left child of the node
Old left child lose right child, and node became new right child of the new root

Insertion

Insertion of new node
If node less that current node we need to insert in left child
If node value bigger that current node we need to insert in right child
After insertion of a new node, we should restore balance. Method of balancing works from new node to root for all subtrees by recursion.

AVL trees have a balancing factor. This calculates the difference between height of a right subtree and height of a left subtree. There are 4 cases for balancing:
If we have a balance factor equal 2, this means the right subtree is too long, we have to apply left rotation.

If we have a balance factor equal 2, and the balance factor of our right child is negative, we have to apply right rotation for our right child and after left rotation for the current vertex.

Same procedure for left subtree.
If we have a balance factor equal -2, this means our left subtree is too long. We have to apply the right rotation.

If we have a balance factor equal -2, and the balance factor of our left child is positive, we have to apply left rotation for our left child and after right rotation for the current vertex.

Removing

If the current node less that value we will look in the left subtree.
If the current node is bigger that value we will look in the right subtree.
There are two cases here:
Node has no right subtree, we have to replace the current vertex to the left child.

Node has a right subtree. We have to find the minimum value in the right subtree, to change the current vertex. After changing vertices we need to remove a minimum value from the right subtree.

Code for tree

class AvlTree {
    constructor() {
        this.root = undefined;
    }

    /**
     * Left rotation
     * 1) Right child became new root
     * 2) Node lose right child, and left child of old right child became new right child of the node
     * 3) Old right child lose left child, and node became new left child of the new root
     * @param {object} node
     * @returns {object}
     */
    rotateLeft(node) {
        const root = node.right;

        const left = node;
        left.right = root.left;
        root.left = left;

        left.fixHeight();
        root.fixHeight();

        return root;
    }

    /**
     * Right rotation
     * 1) Left child became new root
     * 2) Node lose left child, and right child of old left child became new left child of the node
     * 3) Old left child lose right child, and node became new right child of the new root
     * @param {object} node
     * @returns {object}
     */
    rotateRight(node) {
        const root = node.left;

        const right = node;
        right.left = root.right;
        root.right = right;

        right.fixHeight();
        root.fixHeight();

        return root;
    }

    /**
     * AVL trees have a balancing factor. This calculates the difference between height of a right subtree and height of a left subtree.
     * There are 4 cases for balancing.
     * If we have a balance factor equal 2, this means the right subtree is too long, we have to apply left rotation.
     * If we have a balance factor equal 2, and the balance factor of our right child is negative.
     * We have to apply right rotation for our right child and after left rotation for the current vertex.
     * Same procedure for left subtree.
     * If we have a balance factor equal -2, this means our left subtree is too long. We have to apply the right rotation.
     * If we have a balance factor equal -2, and the balance factor of our left child is positive.
     * We have to apply left rotation for our left child and after right rotation for the current vertex.
     * @param {object} node
     * @returns {object}
     */
    balance(node) {
        node.fixHeight();

        if (node.getBalanceFactor() === 2 ) {
            if (node.right.getBalanceFactor() < 0) {
                node.right = this.rotateRight(node.right);
            }
            return this.rotateLeft(node);
        }

        if (node.getBalanceFactor() === -2) {
            if (node.left.getBalanceFactor() > 0) {
                node.left = this.rotateLeft(node.left);
            }
            return this.rotateRight(node);
        }

        return node;
    }

    insert(node) {
        if (! this.root) {
            this.root = node;
            return;
        }
        this.root = this._insert(this.root, node);
    }

    /**
     * Insertion of new node
     * If node less that current node we need to insert in left child
     * If node value bigger that current node we need to insert in right child
     * After insertion of a new node, we should restore balance. Method of balancing works from new node to root for all subtrees by recursion.
     * @param {object} vertex - vertex of the tree
     * @param {object} node - new node
     */
    _insert(vertex, node) {
        if (node.key === vertex.key) {
            return vertex;
        }
        if (node.key < vertex.key) {
            if (! vertex.left) {
                vertex.left = node;
            } else {
                vertex.left = this._insert(vertex.left, node);
            }
        } else {
            if (! vertex.right) {
                vertex.right = node;
            } else {
                vertex.right = this._insert(vertex.right, node);
            }
        }

        return this.balance(vertex);
    }

    findMin(node) {
        return node.left ? this.findMin(node.left) : node;
    }

    /**
     * Removing a minimum in subtree.
     * If a vertex has no left child, we replace the current vertex by the right child.
     *
     * If a vertex has a left child we apply a function for the left child.
     * If a left child has a right child this will be replaced. If a left child has no children this will be replaced by undefined, removed.
     *
     * @param {object} node
     * @returns {object}
     */
    removeMin(node) {
        if (! node.left) {
            return node.right;
        }
        node.left = this.removeMin(node.left);
        return this.balance(node);
    }

    remove(k) {
        this.root = this._remove(this.root, k);
        return this.root;
    }

    /**
     * Removing a node. This is a recursive function.
     * If the current node less that value we will look in the left subtree.
     * If the current node is bigger that value we will look in the right subtree.
     *
     * There are two cases here:
     * Node has no right subtree, we have to replace the current vertex to the left child.
     *
     * Node has a right subtree. We have to find the minimum value in the right subtree, to change the current vertex.
     * After changing vertices we need to remove a minimum value from the right subtree.
     * @param {object} node - current vertex of the tree
     * @param {number} k - value for remove
     * @returns {object|undefined}
     */
    _remove(node, k) {
        if (! node) {
            return;
        }

        if (k < node.key) {
            node.left = this._remove(node.left, k);
        } else if (k > node.key) {
            node.right = this._remove(node.right, k);
        } else {
            if (! node.right) {
                return node.left;
            }

            const min = this.findMin(node.right);
            node.key = min.key;
            node.right = this.removeMin(node.right);

            node = this.balance(node);
        }

        return node;
    }

    find(k, node) {
        if (! node) {
            node = this.root;
        }

        if (k === node.key) {
            return node;
        } else if (k < node.key) {
            if (! node.left) {
                return;
            }
            return this.find(k, node.left);
        } else if (k > node.key) {
            if (! node.right) {
                return;
            }
            return this.find(k, node.right);
        }
    }
}

Example

const printTree = (binaryTree) => {
    const layers = [];

    const queue = [
        {
            node: binaryTree.root,
            layer: 0,
        }
    ];


    while (queue.length) {
        const { node, layer, parent } = queue.shift();

        layers[layer] = layers[layer] || [];
        layers[layer].push(`${node.key}(${parent || ''})`);

        if (node.left) {
            queue.push({
                node: node.left,
                layer: layer + 1,
                parent: node.key + ':l',
            });
        }

        if (node.right) {
            queue.push({
                node: node.right,
                layer: layer + 1,
                parent: node.key + ':r'
            });
        }
    }

    for (let i = 0; i < layers.length; ++i) {
        const layer = layers[i];
        const tabs = new Array(layers.length - i).fill('    ').join('');
        const str = layer.join('    ');
        console.log(tabs + str + '\n');
    }

}

const main = () => {
    const avl = new AvlTree();

    for (let i = 1; i < 11; ++i) {
        const n = new Node({ key: i });
        avl.insert(n);
    }

    printTree(avl);

    avl.insert(new Node({key: 11}));
    printTree(avl);

    avl.remove(10);
    printTree(avl);

};

main();

Golang web application for beginners

Igor Dubinin — Wed, 21 Apr 2021 19:29:12 +0000

Code of example

Web api with golang for beginners

Go is an open source programming language that makes it easy to build simple, reliable, and efficient software etc. I want to build web api with golang, because big part of my work is backend of web or mobile applications.

GVM

First step for work with golang is install golang.
Gvm - golang version manager, it's util to install and use different versions of golang.

bash < <(curl -s -S -L https://raw.githubusercontent.com/moovweb/gvm/master/binscripts/gvm-installer)

gvm install go1.15.6 -B
gvm use go1.15.6 [--default]
gvm list

Go modules

If i want to make workable project with any language i should install libraries from other developers and organizations. Golang have stuff called Go Modules - a module is a collection of Go packages stored in a file tree with a go.mod file at its root.

go mod init mini-seller

Viper

When i develop programing product i should have configuration for web server, databases, file storage, api for integrations. Viper is comfortable and powerful package for management of configuration.

go get github.com/spf13/viper

Mongo

To save data for my project i will use Mongo it's nosql database with simple syntax, big possibilities and good performance.

go get go.mongodb.org/mongo-driver

Testing

When i develop something bigger than landing page i should know that my functionality is working. Launch tests in golang is very simple.

go test -v ./...

Gqlgen

And one of important things for web application is itself web application. I select gqlgen for this, it have generation of code and big possibilities to make graphql api.

go get github.com/99designs/gqlgen

Desing

After installation of packages need to plane design of application. Building a structure of big application is not simple, and before do it, very helpful to read about solid rules, clean architecture, n-tier architecture. Very shortly central part of application is entities of business logic. Business logic of project builds around entities. Logic should not depend from web frameworks or data storage like orm or web api. Use cases shouldn't receive data from storage but should use classes of repository for this. Repositories will help to change one database to another or message broker or web api. For testing of business logic uses mocks of repositories. If you have big difficult validation for your logic you should put this in classes of specifications. And you should not relate your logic with web framework, instead of this web application should depended from use cases, this make changing of web framework more simple.

In my case i made folders:

application - code for web application
1. server - web server for project
2. graph - folder with gqlgen application
  1. schemas - folder contain schemas for graphql
  2. model - folder contain entities for graphql
  3. resolvers - folder contain controllers for graphql
domain - business logic of project
1. common - common data for all packages
  1. entities - here i save all entities, these could be entities for business logic, models for database and dto for communication between classes
2. packages - here logic of project
  1. catalogpkg - logic for catalog of products
    1. usecase - interface for use case, it describe functionality available in package
    2. repository - interface describe functionality of data storage
    3. catalogusecase - contain use case - business logic of catalog and test for use case
    4. catalogrepository - contain repository with requests for database, integration tests for validation of repository, and mock of repository, it needed to test logic of use case without real connection to database
infrastructure - code without business logic, database driver, helper for configuration
1. mongohelper - contain functionality for configure connection to mongo database and folder with test data. These files need to initialize empty project with demo data, and i use these for testing of logic.
2. viperhelper - i use this helper to read configuration for project and for test. This helps me to change configuration from default to local and use environment variables.

Infrastructure

Viper helper read local configuration for project and tests, and can use environment variables instead of configuration files.

// Package viperhelper - helper for viper, uses to apply configuration from local files and environment
package viperhelper

import (
    "os"

    "github.com/prometheus/common/log"
    "github.com/spf13/viper"
)

// IViper - helper for viper, describe methods for initiation of config
type IViper interface {
    updateSettings()
    Read() error
    getEnv()
}

// default values for json config
const (
    configType = "json"
    configName = "config.json"
    configPath = "./infrastructure/viperhelper"
)

// list of environment variables
var envVariables = []string{
    "MONGO_DB", "MONGO_HOST", "MONGO_PORT", "MONGO_USER",
    "MONGO_PASSWORD", "MONGO_AUTH", "MONGO_REPLICASET",
    "WEB_PORT",
}

// Viper - class for initialization of viper config with values from local configuration and environment
type Viper struct {
    ConfigType, ConfigName, ConfigPath string
}

// updateSettings - set default values for arguments
func (vip *Viper) updateSettings() {
    if vip.ConfigType == "" {
        vip.ConfigType = configType
    }
    if vip.ConfigName == "" {
        vip.ConfigName = configName
    }
    if vip.ConfigPath == "" {
        vip.ConfigPath = configPath
    }
}

// getEnv - get variables from environment
func (vip *Viper) getEnv() {
    for _, variable := range envVariables {
        value := os.Getenv(variable)
        if value != "" {
            viper.Set(variable, value)
        }
    }
}

// Read - read configuration
func (vip *Viper) Read() error {
    vip.updateSettings()

    viper.SetConfigType(vip.ConfigType)
    viper.AddConfigPath(vip.ConfigPath)

    viper.SetConfigName("local-" + vip.ConfigName)
    err := viper.ReadInConfig()
    if err != nil {
        log.Info("Local config", err)
        viper.SetConfigName(vip.ConfigName)
        err := viper.ReadInConfig()
        if err != nil {
            return err
        }
    }

    vip.getEnv()

    return nil
}

Mongo helper take configuration from viper and connect to database, with authentication and replicaset.

package mongohelper

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/spf13/viper"
    "go.mongodb.org/mongo-driver/mongo"
    "go.mongodb.org/mongo-driver/mongo/options"
)

// Connect connection to mongodb
func Connect(dbname string) (*mongo.Database, error) {
    fmt.Println("init mongodb")
    if dbname == "" {
        dbname = viper.GetString("MONGO_DB")
    }

    // connection string
    uri := "mongodb://"
    if viper.GetBool("MONGO_AUTH") {
        uri += viper.GetString("MONGO_USER") + ":" + viper.GetString("MONGO_PASSWORD") + "@"
    }
    uri += viper.GetString("MONGO_HOST") + ":" + viper.GetString("MONGO_PORT")

    replicaSet := viper.GetString("MONGO_REPLICASET")
    if replicaSet != "" {
        uri += "/" + dbname + "?replicaSet=" + replicaSet
    }

    // create connection
    client, err := mongo.NewClient(options.Client().ApplyURI(uri))
    if err != nil {
        log.Fatalf("err")
        return nil, err
    }

    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    err = client.Connect(ctx)
    if err != nil {
        log.Fatal(err)
        return nil, err
    }

    err = client.Ping(context.Background(), nil)
    if err != nil {
        log.Fatal(err)
        return nil, err
    }

    return client.Database(dbname), nil
}

Test data - mongodb helper, contain folder with default data for project. Data helper give possibility read default data, and insert this data to database. This helpful for launching empty project and testing.

package testdata

import (
    "context"
    "encoding/json"
    "io/ioutil"
    "log"
    "mini-seller/domain/common/entities/employeeentity"
    "mini-seller/domain/common/entities/organizationentity"
    "mini-seller/domain/common/entities/productcategoryentity"
    "mini-seller/domain/common/entities/productentity"
    "time"

    "go.mongodb.org/mongo-driver/bson"
    "go.mongodb.org/mongo-driver/mongo"
)

// ProductsData - dto for products test data
type ProductsData struct {
    Categories []productcategoryentity.ProductCategory
    Products   []productentity.Product
}

// GetProducts - read json of products
func GetProducts() (*ProductsData, error) {
    content, err := ioutil.ReadFile("infrastructure/mongohelper/testdata/products.json")
    if err != nil {
        log.Fatal("Error when opening file: ", err)
    }

    var productData ProductsData
    err = json.Unmarshal(content, &productData)
    if err != nil {
        log.Fatal("Unmarshal: ", err)
        return nil, err
    }

    return &productData, nil
}

// InsertProducts - insert test data for categories and products
func InsertProducts(db *mongo.Database) {
    productData, err := GetProducts()
    if err != nil {
        log.Fatal("Unmarshal: ", err)
    }

    categories := make([]interface{}, len(productData.Categories))
    for i, cat := range productData.Categories {
        entity := productcategoryentity.ToMongo(&cat)
        categories[i] = entity
    }

    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    _, err = db.Collection("product_categories").DeleteMany(ctx, bson.M{})
    if err != nil {
        log.Fatal(err)
    }
    _, err = db.Collection("product_categories").InsertMany(ctx, categories)
    if err != nil {
        log.Fatal(err)
    }

    products := make([]interface{}, len(productData.Products))
    for i, product := range productData.Products {
        entity := productentity.ToProductMongo(&product)
        products[i] = entity
    }

    _, err = db.Collection("products").DeleteMany(ctx, bson.M{})
    if err != nil {
        log.Fatal(err)
    }
    _, err = db.Collection("products").InsertMany(ctx, products)
    if err != nil {
        log.Fatal(err)
    }
}

Entities

Entities this is human understandable entities of business logic like product or customer.

package productentity

// Product - entity of product
type Product struct {
    ID             string
    IDCategory     string
    IDOrganization string
    Name           string
    Price          int
    Count          int
    Status         string
}

package productcategoryentity

// ProductCategory is the entity of category
type ProductCategory struct {
    ID     string
    Name   string
    Status string
}

package organizationentity

// Organization is the entity of organization
type Organization struct {
    ID          string   `validate:"omitempty"`
    Name        string   `validate:"required"`
    Email       string   `validate:"required"`
    Phone       string   `validate:"required"`
    Status      string   `validate:"required"`
    IDsCategory []string `validate:"required"`
}

Models

Models contain realization of entity for saving in database and methods to convert entity to model. At example shows model for product.

package productentity

import (
    "go.mongodb.org/mongo-driver/bson/primitive"
)

// ProductMongo - model for mongo database
type ProductMongo struct {
    ID             primitive.ObjectID `bson:"_id,omitempty"`
    IDCategory     primitive.ObjectID `bson:"_id_cat"`
    IDOrganization primitive.ObjectID `bson:"_id_org"`
    Name           string
    Price          int
    Count          int
    Status         string
}

// ToProduct - method to convert model to entity
func ToProduct(pm *ProductMongo) *Product {
    ID := ""
    if !pm.ID.IsZero() {
        ID = pm.ID.Hex()
    }

    return &Product{
        ID:             ID,
        IDCategory:     pm.IDCategory.Hex(),
        IDOrganization: pm.IDOrganization.Hex(),
        Name:           pm.Name,
        Price:          pm.Price,
        Count:          pm.Count,
        Status:         pm.Status,
    }
}

// ToProductMongo - method to convert entity to model
func ToProductMongo(p *Product) *ProductMongo {
    var ID primitive.ObjectID
    if p.ID != "" {
        ID, _ = primitive.ObjectIDFromHex(p.ID)
    }
    idCategory, _ := primitive.ObjectIDFromHex(p.IDCategory)
    idOrganization, _ := primitive.ObjectIDFromHex(p.IDOrganization)

    return &ProductMongo{
        ID:             ID,
        IDCategory:     idCategory,
        IDOrganization: idOrganization,
        Name:           p.Name,
        Price:          p.Price,
        Count:          p.Count,
        Status:         p.Status,
    }
}

Interfaces

Interfaces describe classes. Communication between packages should be abstract and builds by interfaces. It help change realisation of methods and apply test classes instead real.

package catalogpkg

import (
    "context"
    "mini-seller/domain/common/entities/organizationentity"
    "mini-seller/domain/common/entities/productcategoryentity"
    "mini-seller/domain/common/entities/productentity"
)

type IUseCase interface {
    GetOrganizationList(ctx context.Context) ([]*organizationentity.Organization, error)
    GetOrganizationDetail(ctx context.Context, id string) (*organizationentity.Organization, error)

    GetCategoryList(ctx context.Context, ids []string) ([]*productcategoryentity.ProductCategory, error)
    GetCategoryDetail(ctx context.Context, id string) (*productcategoryentity.ProductCategory, error)

    GetProductList(ctx context.Context, ids_organization []string, ids_category []string) ([]*productentity.Product, error)
    GetProductDetail(ctx context.Context, id string) (*productentity.Product, error)
}

type IRepository interface {
    GetOrganizationList(ctx context.Context) ([]*organizationentity.Organization, error)
    GetOrganizationDetail(ctx context.Context, id string) (*organizationentity.Organization, error)

    GetCategoryList(ctx context.Context, ids []string) ([]*productcategoryentity.ProductCategory, error)
    GetCategoryDetail(ctx context.Context, id string) (*productcategoryentity.ProductCategory, error)

    GetProductList(ctx context.Context, ids_organization []string, ids_category []string) ([]*productentity.Product, error)
    GetProductDetail(ctx context.Context, id string) (*productentity.Product, error)
}

Use case

Use cases contain logic of application. Example contain simple validation and calling of repositories for api of products.

package catalogusecase

import (
    "context"
    "mini-seller/domain/common/entities/organizationentity"
    "mini-seller/domain/common/entities/productcategoryentity"
    "mini-seller/domain/common/entities/productentity"
    "mini-seller/domain/packages/catalogpkg"
)

type UseCase struct {
    catalogRepo catalogpkg.IRepository
}

func NewCatalogUseCase(catalogRepo catalogpkg.IRepository) *UseCase {
    return &UseCase{catalogRepo: catalogRepo}
}

func (cUseCase UseCase) GetProductList(ctx context.Context, ids_organization []string, ids_category []string) ([]*productentity.Product, error) {
    return cUseCase.catalogRepo.GetProductList(ctx, ids_organization, ids_category)
}
func (cUseCase UseCase) GetProductDetail(ctx context.Context, id string) (*productentity.Product, error) {
    if id == "" {
        return nil, catalogpkg.ErrProductNotFound
    }
    return cUseCase.catalogRepo.GetProductDetail(ctx, id)
}

Test for use case

Tests contain verification of use case behavior with valid data and invalid data. Example for products.

package catalogusecase

import (
    "context"
    "os"
    "testing"

    "mini-seller/domain/packages/catalogpkg"
    "mini-seller/domain/packages/catalogpkg/catalogrepository"

    "github.com/stretchr/testify/assert"
    "go.mongodb.org/mongo-driver/mongo"
)

var db *mongo.Database

// TestMain - integration tests for repository
func TestMain(m *testing.M) {
    os.Chdir("../../../../")
    retCode := m.Run()
    os.Exit(retCode)
}

func TestGetProductList(t *testing.T) {
    t.Log("Test catalogusecase products list")

    catalogRepo := catalogrepository.NewCatalogRepositoryMock()
    catalogUseCase := NewCatalogUseCase(catalogRepo)

    prodList, err := catalogUseCase.GetProductList(context.TODO(), nil, []string{"604488100f719d9c76a28fe3"})
    assert.Nil(t, err)
    assert.Equal(t, len(prodList), 2)
    assert.Equal(t, prodList[0].Name, "Cola")

    prodList, err = catalogUseCase.GetProductList(context.TODO(), []string{"6043d76e94df8de741c2c0d5"}, nil)
    assert.Nil(t, err)
    assert.Equal(t, len(prodList), 4)
    assert.Equal(t, prodList[0].Name, "Cola")

    prodList, err = catalogUseCase.GetProductList(context.TODO(), []string{"6043d76e94df8de741c2c0d5"}, []string{"604488100f719d9c76a28fe7"})
    assert.Nil(t, err)
    assert.Equal(t, len(prodList), 3)
    assert.Equal(t, prodList[0].Name, "Chicken Barbecue")
}

func TestGetProductDetail(t *testing.T) {
    t.Log("Test catalogusecase product detail")

    catalogRepo := catalogrepository.NewCatalogRepositoryMock()
    catalogUseCase := NewCatalogUseCase(catalogRepo)

    product, err := catalogUseCase.GetProductDetail(context.TODO(), "604497558ffcad558eb8e1f4")
    assert.Nil(t, err)
    assert.Equal(t, product.Name, "Salad Cesar")

    product, err = catalogUseCase.GetProductDetail(context.TODO(), "")
    assert.Equal(t, err, catalogpkg.ErrProductNotFound)
}

Repository

Repositories contain methods for management of data. Example for search products in mongodb.

package catalogrepository

import (
    "context"
    "mini-seller/domain/common/entities/productentity"
    "mini-seller/domain/packages/catalogpkg"
    "time"

    "go.mongodb.org/mongo-driver/mongo"
    "go.mongodb.org/mongo-driver/bson"
    "go.mongodb.org/mongo-driver/bson/primitive"
    "go.mongodb.org/mongo-driver/mongo/options"
)

type Repository struct {
    db *mongo.Database
}

// NewCatalogRepository - constructor for catalog repository
func NewCatalogRepository(db *mongo.Database) *Repository {
    return &Repository{
        db: db,
    }
}

func (cRepo Repository) GetProductList(contx context.Context, ids_organization []string, ids_category []string) ([]*productentity.Product, error) {
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    // query
    query := bson.M{"status": catalogpkg.StatusActive}

    if ids_organization != nil && len(ids_organization) > 0 {
        ids_org := make([]primitive.ObjectID, len(ids_organization))
        for i, id := range ids_organization {
            id_org, err_id := primitive.ObjectIDFromHex(id)
            if err_id != nil {
                return nil, err_id
            }
            ids_org[i] = id_org
        }
        query["_id_org"] = bson.M{"$in": ids_org}
    }

    if ids_category != nil && len(ids_category) > 0 {
        ids_cat := make([]primitive.ObjectID, len(ids_category))
        for i, id := range ids_category {
            id_cat, err_id := primitive.ObjectIDFromHex(id)
            if err_id != nil {
                return nil, err_id
            }
            ids_cat[i] = id_cat
        }
        query["_id_cat"] = bson.M{"$in": ids_cat}
    }

    // sort
    findOptions := options.Find()
    findOptions.SetSort(bson.D{{"_id", 1}})
    // request
    cursor, err := cRepo.db.Collection("products").Find(ctx, query, findOptions)
    if err != nil {
        return nil, err
    }

    // format data from cursor
    products := make([]*productentity.Product, 0)
    defer cursor.Close(ctx)

    for cursor.Next(ctx) {
        prodMongo := productentity.ProductMongo{}
        err := cursor.Decode(&prodMongo)
        if err != nil {
            return nil, err
        }
        product := productentity.ToProduct(&prodMongo)
        products = append(products, product)
    }

    // answer
    return products, nil
}

func (cRepo Repository) GetProductDetail(contx context.Context, id string) (*productentity.Product, error) {
    // convert id to bson
    ID, err := primitive.ObjectIDFromHex(id)
    if err != nil {
        return nil, err
    }

    // get context
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    // context
    prodMongo := productentity.ProductMongo{}
    err = cRepo.db.Collection("products").FindOne(ctx, bson.M{"_id": ID, "status": catalogpkg.StatusActive}).Decode(&prodMongo)
    if err != nil {
        return nil, err
    }

    // convert to entity
    product := productentity.ToProduct(&prodMongo)

    // answer
    return product, nil
}

Tests for Repository

Tests of repositories should check that database queries in repository is valid. To check this, i insert default data into test database, launch repository and compare answer with correct result.

package catalogrepository

import (
    "context"
    "log"
    "mini-seller/infrastructure/mongohelper"
    "mini-seller/infrastructure/mongohelper/testdata"
    "mini-seller/infrastructure/viperhelper"
    "os"
    "testing"

    "github.com/stretchr/testify/assert"
    "go.mongodb.org/mongo-driver/mongo"
)

var db *mongo.Database

// TestMain - integration tests for repository
func TestMain(m *testing.M) {
    setUp()
    retCode := m.Run()
    tearDown()
    os.Exit(retCode)
}

func setUp() {
    os.Chdir("../../../../")
    vip := viperhelper.Viper{ConfigType: "", ConfigName: "test.json", ConfigPath: "infrastructure/viperhelper"}
    err := vip.Read()
    if err != nil {
        log.Fatal("Catalog", err)
    }

    db, err = mongohelper.Connect("test_catalog")
    if err != nil {
        log.Fatal(err)
    }

    testdata.InsertOrganizations(db)
    testdata.InsertProducts(db)
}

func tearDown() {
    db.Drop(context.TODO())
}

func TestGetProductList(t *testing.T) {
    t.Log("Test catalogrepository products list")

    catalogRepo := NewCatalogRepository(db)

    prodList, err := catalogRepo.GetProductList(context.TODO(), nil, []string{"604488100f719d9c76a28fe3"})
    assert.Nil(t, err)
    assert.Equal(t, len(prodList), 2)
    assert.Equal(t, prodList[0].Name, "Cola")

    prodList, err = catalogRepo.GetProductList(context.TODO(), []string{"6043d76e94df8de741c2c0d5"}, nil)
    assert.Nil(t, err)
    assert.Equal(t, len(prodList), 4)
    assert.Equal(t, prodList[0].Name, "Cola")

    prodList, err = catalogRepo.GetProductList(context.TODO(), []string{"6043d76e94df8de741c2c0d5"}, []string{"604488100f719d9c76a28fe7"})
    assert.Nil(t, err)
    assert.Equal(t, len(prodList), 3)
    assert.Equal(t, prodList[0].Name, "Chicken Barbecue")
}

func TestGetProductDetail(t *testing.T) {
    t.Log("Test catalogrepository product detail")

    catalogRepo := NewCatalogRepository(db)

    product, err := catalogRepo.GetProductDetail(context.TODO(), "604497558ffcad558eb8e1f4")
    assert.Nil(t, err)
    assert.Equal(t, product.Name, "Salad Cesar")
}

Mock for Repository

Mock of repository is necessary for testing of use cases. Business logic should not depend from data source, it works with entities.

package catalogrepository

import (
    "context"
    "mini-seller/domain/common/entities/productentity"
    "mini-seller/infrastructure/mongohelper/testdata"
)

// RepositoryMock - mock instead database
type RepositoryMock struct {
    products      []*productentity.Product
}

// NewCatalogRepositoryMock - constructor for mock of repository with fixed data
func NewCatalogRepositoryMock() *RepositoryMock {
    prodData, _ := testdata.GetProducts()

    products := make([]*productentity.Product, len(prodData.Products))
    for i := range prodData.Products {
        products[i] = &prodData.Products[i]
    }
    categories := make([]*productcategoryentity.ProductCategory, len(prodData.Categories))
    for i := range prodData.Categories {
        categories[i] = &prodData.Categories[i]
    }

    return &RepositoryMock{
        organizations: organizations,
        products:      products,
        categories:    categories,
    }
}

func includes(arr []string, val string) bool {
    for _, str := range arr {
        if str == val {
            return true
        }
    }
    return false
}

func prodQuery(ids_organization []string, ids_category []string, prod *productentity.Product) bool {
    var byOrg bool = true
    var byCat bool = true

    if ids_organization != nil {
        byOrg = includes(ids_organization, prod.IDOrganization)
    }

    if ids_category != nil {
        byCat = includes(ids_category, prod.IDCategory)
    }

    return byOrg && byCat
}

func (r *RepositoryMock) GetProductList(ctx context.Context, ids_organization []string, ids_category []string) ([]*productentity.Product, error) {
    products := make([]*productentity.Product, 0)

    for _, prod := range r.products {
        if prodQuery(ids_organization, ids_category, prod) {
            products = append(products, prod)
        }
    }

    return products, nil
}

func (r *RepositoryMock) GetProductDetail(ctx context.Context, id string) (*productentity.Product, error) {
    for _, prod := range r.products {
        if id == prod.ID {
            return prod, nil
        }
    }
    return nil, nil
}

Gqlgen

Initialization of gqlgen with recommended folder structure by running this command

go run github.com/99designs/gqlgen init

My structure of folders:

schemas
- catalog.graphqls
- query.graphqls
model
- catalog
resolvers
- catalog.resolvers.go
- query.resolvers.go
- catalogdataloader.go
- resolver.go
generated

Gqlgen can generate models and code for resolvers by schemas

go run github.com/99designs/gqlgen generate

To run go generate recursively over your entire project, use this command:

go generate ./...

Schemas

# entities
type Organization {
  id: String!
  name: String!
  email: String!
  phone: String!
  status: String!
  categories: [Category]
  products: [Product]
}

type Category {
  id: String!
  name: String!
  status: String!
  products: [Product]
}

type Product {
  id: String!
  name: String!
  status: String!
  price: Int!
  count: Int!
  category: Category
  organization: Organization
}

# query
type Query {
  organizations: [Organization!]!
  product(id: String!): Product
}

Models

package model

// Product - entity for product
type Product struct {
    ID             string `json:"id"`
    IDCategory     string `json:"category"`
    IDOrganization string `json:"organization"`
    Name           string `json:"name"`
    Price          int    `json:"price"`
    Count          int    `json:"count"`
    Status         string `json:"status"`
}

// Category - entity for category of products
type Category struct {
    ID     string `json:"id"`
    Name   string `json:"name"`
    Status string `json:"status"`
    IDOrg  string `json:"idorg"`
}

// Organization - entity for organization
type Organization struct {
    ID          string   `json:"id"`
    Name        string   `json:"name"`
    Email       string   `json:"email"`
    Phone       string   `json:"phone"`
    Status      string   `json:"status"`
    IDsCategory []string `json:"categories"`
}

Resolvers

Resolver

package resolvers

//go:generate go run github.com/99designs/gqlgen

import (
    "mini-seller/application/graph/model"
    "mini-seller/domain/packages/catalogpkg"
)

// This file will not be regenerated automatically.
//
// It serves as dependency injection for your app, add any dependencies you require here.

type Resolver struct {
    CatalogUseCase catalogpkg.IUseCase

    organizations []*model.Organization
    product       *model.Product
}

Main resolver for query

package resolvers

// This file will be automatically regenerated based on the schema, any resolver implementations
// will be copied through when generating and any unknown code will be moved to the end.

import (
    "context"
    "mini-seller/application/graph/generated"
    "mini-seller/application/graph/model"

    "github.com/prometheus/common/log"
)

func (r *queryResolver) Organizations(ctx context.Context) ([]*model.Organization, error) {
    orgs, err := r.CatalogUseCase.GetOrganizationList(ctx)
    if err != nil {
        log.Warn(err)
        return nil, err
    }
    if len(orgs) == 0 {
        return nil, nil
    }

    organizations := make([]*model.Organization, len(orgs))
    for i, org := range orgs {
        organizations[i] = &model.Organization{
            ID:          org.ID,
            Name:        org.Name,
            Email:       org.Email,
            Phone:       org.Phone,
            Status:      org.Status,
            IDsCategory: org.IDsCategory,
        }
    }

    return organizations, err
}

func (r *queryResolver) Product(ctx context.Context, id string) (*model.Product, error) {
    prod, err := r.CatalogUseCase.GetProductDetail(ctx, id)
    if err != nil {
        log.Warn(err)
        return nil, err
    }
    if prod == nil {
        return nil, nil
    }

    product := &model.Product{
        ID:             prod.ID,
        IDCategory:     prod.IDCategory,
        IDOrganization: prod.IDOrganization,
        Name:           prod.Name,
        Price:          prod.Price,
        Count:          prod.Count,
        Status:         prod.Status,
    }

    return product, nil
}

// Query returns generated.QueryResolver implementation.
func (r *Resolver) Query() generated.QueryResolver { return &queryResolver{r} }

type queryResolver struct{ *Resolver }

Resolver for entities

package resolvers

// This file will be automatically regenerated based on the schema, any resolver implementations
// will be copied through when generating and any unknown code will be moved to the end.

import (
    "context"
    "mini-seller/application/graph/generated"
    "mini-seller/application/graph/model"

    "github.com/prometheus/common/log"
)

func (r *categoryResolver) Products(ctx context.Context, obj *model.Category) ([]*model.Product, error) {
    IDOrg := ""
    if obj.IDOrg != "" {
        IDOrg = obj.IDOrg
    }

    prodList, err := r.CatalogUseCase.GetProductList(ctx, []string{IDOrg}, []string{obj.ID})
    if err != nil {
        log.Warn(err)
        return nil, err
    }
    if prodList == nil {
        return nil, nil
    }

    products := make([]*model.Product, len(prodList))
    for i, product := range prodList {
        products[i] = &model.Product{
            ID:             product.ID,
            IDCategory:     product.IDCategory,
            IDOrganization: product.IDOrganization,
            Name:           product.Name,
            Price:          product.Price,
            Count:          product.Count,
            Status:         product.Status,
        }
    }

    return products, nil
}

func (r *organizationResolver) Categories(ctx context.Context, obj *model.Organization) ([]*model.Category, error) {
    catList, err := r.CatalogUseCase.GetCategoryList(ctx, obj.IDsCategory)
    if err != nil {
        log.Warn(err)
        return nil, err
    }
    if catList == nil {
        return nil, nil
    }

    categories := make([]*model.Category, len(catList))
    for i, category := range catList {
        categories[i] = &model.Category{
            ID:     category.ID,
            Name:   category.Name,
            Status: category.Status,
            IDOrg:  obj.ID,
        }
    }

    return categories, nil
}

func (r *organizationResolver) Products(ctx context.Context, obj *model.Organization) ([]*model.Product, error) {
    return ctxLoaders(ctx).productsByOrganization.Load(obj.ID)
}

func (r *productResolver) Category(ctx context.Context, obj *model.Product) (*model.Category, error) {
    cat, err := r.CatalogUseCase.GetCategoryDetail(ctx, obj.IDCategory)
    if err != nil {
        return nil, err
    }
    if cat == nil {
        return nil, nil
    }

    category := &model.Category{
        ID:     cat.ID,
        Name:   cat.Name,
        Status: cat.Status,
        IDOrg:  obj.IDOrganization,
    }

    return category, nil
}

func (r *productResolver) Organization(ctx context.Context, obj *model.Product) (*model.Organization, error) {
    org, err := r.CatalogUseCase.GetOrganizationDetail(ctx, obj.IDOrganization)
    if err != nil {
        return nil, err
    }
    if org == nil {
        return nil, nil
    }

    organization := &model.Organization{
        ID:          org.ID,
        Name:        org.Name,
        Email:       org.Email,
        Phone:       org.Phone,
        Status:      org.Status,
        IDsCategory: org.IDsCategory,
    }

    return organization, nil
}

// Category returns generated.CategoryResolver implementation.
func (r *Resolver) Category() generated.CategoryResolver { return &categoryResolver{r} }

// Organization returns generated.OrganizationResolver implementation.
func (r *Resolver) Organization() generated.OrganizationResolver { return &organizationResolver{r} }

// Product returns generated.ProductResolver implementation.
func (r *Resolver) Product() generated.ProductResolver { return &productResolver{r} }

type categoryResolver struct{ *Resolver }
type organizationResolver struct{ *Resolver }
type productResolver struct{ *Resolver }

Web server

Launching of web api:

Reading of configuration
Connection to database
Initialization of repositories
Initialization of use cases
Initialization of web framework
Launching of web server

package main

import (
    "log"
    "mini-seller/application/graph/generated"
    "mini-seller/application/graph/resolvers"
    "mini-seller/domain/packages/catalogpkg/catalogrepository"
    "mini-seller/domain/packages/catalogpkg/catalogusecase"
    "mini-seller/infrastructure/mongohelper"
    "mini-seller/infrastructure/viperhelper"
    "net/http"

    "github.com/99designs/gqlgen/graphql/handler"
    "github.com/99designs/gqlgen/graphql/playground"
    "github.com/spf13/viper"
)

const defaultPort = "8080"

func main() {
    vip := viperhelper.Viper{ConfigType: "", ConfigName: "", ConfigPath: "infrastructure/viperhelper"}
    vip.Read()

    db, err := mongohelper.Connect("")
    if err != nil {
        log.Fatal(err)
    }

    catalogRepository := catalogrepository.NewCatalogRepository(db)
    catalogUseCase := catalogusecase.NewCatalogUseCase(catalogRepository)

    resolver := resolvers.Resolver{CatalogUseCase: catalogUseCase}

    router := http.NewServeMux()
    router.Handle("/", playground.Handler("GraphQL playground", "/query"))
    router.Handle("/query", handler.NewDefaultServer(generated.NewExecutableSchema(generated.Config{Resolvers: &resolver})))

    port := viper.GetString("WEB_PORT")
    log.Printf("connect to http://localhost:%s/ for GraphQL playground", port)
    log.Fatal(http.ListenAndServe(":"+port, resolvers.LoaderMiddleware(catalogUseCase, router)))
}

Launch server

go run application/server.go

GraphQL

GraphQL is kind of web api, it provide a query language, understandable documentation for your requests and has pretty good playground.

With GraphQL you could select fields that you need from backend and you could create resolvers for related entities. Example for product:

query product($id: String!) {
    product(id: $id) {
        id
        name
        price
        count
        category{
            id
            name
        }
        organization {
            id
            name
        }
    }
}

Graphql have powerful query language, you could use Aliases and get different data in one request and you could use Fragment for common fields.
Query

query products($id_cola: String! $id_salad: String!) {
    cola: product(id: $id_cola) {
        ...detailFields
    }
    salad: product(id: $id_salad) {
        ...detailFields
    }
}

fragment detailFields on Product {
    id
    name
    price
    count
    category{
        id
        name
    }
    organization {
        id
        name
    }
}

Variables

{
    "id_cola": "604497558ffcad558eb8e1f5",
    "id_salad": "604497558ffcad558eb8e1f4"
}

Result

{
    "data": {
        "cola": {
            "id": "604497558ffcad558eb8e1f5",
            "name": "Cola",
            "price": 100,
            "count": 100,
            "category": {
                "id": "604488100f719d9c76a28fe3",
                "name": "Drinks"
            },
            "organization": {
                "id": "6043d76e94df8de741c2c0d6",
                "name": "restaurant"
            }
        },
        "salad": {
            "id": "604497558ffcad558eb8e1f4",
            "name": "Salad Cesar",
            "price": 200,
            "count": 10,
            "category": {
                "id": "604488100f719d9c76a28fe6",
                "name": "Salad"
            },
            "organization": {
                "id": "6043d76e94df8de741c2c0d6",
                "name": "restaurant"
            }
        }
    }
}