DEV Community: Raggi

Dual Node TypeScript Packages - The Easy Way

Raggi — Mon, 29 Jul 2024 21:18:13 +0000

Well we've all been down that road haven't we? trying to build a Node TypeScript package that supports both ESM/CJS without losing our mind in the process.

I took it upon myself to lose my own mind while keeping you sane 🫡

Introducing moshimosh, The easiest way to bootstrap a Node Hybrid TypeScript package with modern tooling that just works, all you need is

  npx moshimosh

You Get

Hybrid ESM/CJS that just works
Tests and Coverage with vitest
Linting/Formatting with Biome
Linting/Formatting Git Pre-commit hooks
Documentation generation from your compiled .d.ts with typedoc
Github action for building and testing
Github action for updating docs in a GitHub Pages repo

And if i saved you some sanity, please check out the GitHub Repo, all contributions are welcome!

The easiest way ever to create a Node TypeScript Project!

Raggi — Sat, 03 Apr 2021 17:45:46 +0000

Hello everyone!

Today I would like to share with you a very simple way to get up and running with a new Node TypeScript project.

All it takes is running a single command from your terminal

$ npx node-ts-new

You would then get a Node.js TypeScript project configured with:

nodemon
eslint
prettier
dotenv

The generated project folder structure looks like this

Thats it!. Please feel free to check out the github repo for this package and even contribute to it!

Next.js Client - Youtube GIF Maker Using Next.js, Node and RabbitMQ

Raggi — Mon, 15 Feb 2021 15:17:38 +0000

Hello everyone,
This Article is the fourth part of the series Youtube GIF Maker Using Next.js, Node and RabbitMQ.

In this article we will dive into building the client side of our Youtube to GIF converter. This Article will contain some code snippets but the whole project can be accessed on github which contains the full source code. You can also view the app demo.

Please note that the code snippets will only include the minimal code required for the functionality (HTML/Code related to styling...etc is ignored)
Also note that Bulma is used for this project but you can use whatever CSS you want.

Functionalities
Home Page
GIF Page
- Polling The GIF Conversion Job
- Server Side Rendering

Functionalities

The Client Side of our app is straight forward, it has to do only two things

Provide an interface for creating GIF Conversion requests from youtube video
Provide a page that keeps polling the GIF conversion job and viewing generated GIF when the job is done

Lets jump straight into building the first one in the home page.

Home Page

Minimally this page has to provide

Input fields containing
- Youtube video url
- GIF start time
- GIF end time
An embedded youtube player showing the selected video as well as showing a preview of the selected time range (start/end times)
Two buttons one for previewing the current selection as well as one for submitting the current selection for generating the GIF

Lets start by creating the three needed input fields and their respective states.

// pages/index.tsx
import React, { useState, useMemo } from 'react';

const Home: React.FC = () => {
  const [youtubeUrl, setYoutubeUrl] = useState("");
  const [startTime, setStartTime] = useState("");
  const [endTime, setEndTime] = useState("");

  const validYoutubeUrl = useMemo(() => {
    const youtubeUrlRegex = /^(?:https?:\/\/)?(?:www\.)?(?:youtu\.be\/|youtube\.com\/(?:embed\/|v\/|watch\?v=|watch\?.+&v=))((\w|-){11})(?:\S+)?$/;
    return youtubeUrl.match(youtubeUrlRegex);
  }, [youtubeUrl]);

  return (
    <>
      <input
       className={`input ${youtubeUrl === "" ? "is-dark" : validYoutubeUrl? "is-success": "is-danger" }`}
       type="text"
       placeholder="Youtube URL, eg: https://www.youtube.com/watch?v=I-QfPUz1es8"
       value={youtubeUrl}
       onChange={(e) => {
         setYoutubeUrl(e.target.value);
       }}
      />

      <input
        className="input is-dark"
        type="number"
        placeholder="Start Second, eg: 38"
        value={startTime}
        onChange={(e) => {
         setStartTime(e.target.value);
        }}
       />

      <input
        className="input is-dark"
        type="number"
        placeholder="End Second, eg: 72"
        value={endTime}
        onChange={(e) => {
         setEndTime(e.target.value);
        }}
      />
    </>
   )
}

Notice that we check for the youtube url validity using Regex. This is not necessary but it is used to provide a good visual feedback as well as will be used to conditionally render the embedded youtube player later on to avoid showing an empty player (can also be ignored).

Now its time to add the embedded youtube player
We will be using the youtube player from react-youtube

// pages/index.tsx
import React, { useState, useMemo } from 'react';
import YouTube from "react-youtube";

const Home: React.FC = () => {
  // ...code from before
  const [ytPlayer, setYtPlayer] = useState(null);
  const ytVideoId = useMemo(() => {
    return youtubeUrl.split("v=")[1]?.slice(0, 11);
   }, [youtubeUrl]);

  return (
    <>
      <div className="content">
         {validYoutubeUrl ? (
           <>
             <h3>Preview</h3>
             <YouTube
               videoId={ytVideoId}
               opts={{
                 playerVars: {
                 start: Number(startTime),
                 end: Number(endTime),
                 autoplay: 0,
                },
               }}
               onReady={(e) => {
                setYtPlayer(e.target);
               }}
             />
            </>
         ) : (
           <h4>No Youtube Video Link Selected</h4>
        )}
      </div>
    </>
   )
}

Notice that we initialized a state ytPlayer with the youtube player event target object. We will use this later to manipulate the player programmatically, specifically when we add the preview button

Now its time to add our two buttons, Preview and Generate

Preview: Used to play the youtube video from the selected start/end times to give the user an idea of how the GIF will look like
Generate: Used to send the actual GIF conversion request. i.e: starting the actual conversion

// pages/index.tsx
import React, { useState } from 'react';
import axios from "axios";
import { useRouter } from "next/router";

const Home: React.FC = () => {
  // ... code from before
  const router = useRouter();
  const [loading, setLoading] = useState(false);
  const submitYoutubeVideo = async () => {
    setLoading(true);
    try {
      const response = await axios.post(
        `${process.env.NEXT_PUBLIC_BASE_URL}/api/v1/jobs`,
        {
          youtubeUrl,
          startTime: Number(startTime),
          endTime: Number(endTime),
        },
        {}
      );
      router.push(`/jobs/${response.data.id}`);
    } catch (err) {
      alert(err?.response?.data?.message || "Something went wrong");
    }
    setLoading(false);
  };

  return (
    <>
     <button
      className="button is-black"
      onClick={() => {
       if (ytPlayer)
         ytPlayer.loadVideoById({
           videoId: ytVideoId,
           startSeconds: Number(startTime),
           endSeconds: Number(endTime),
          });
       }}
      >
       Preview
      </button>

      <button
       className={`button is-black is-outlined ${loading ? "is-loading" : ""}`}
       onClick={submitYoutubeVideo}
       >
        Generate GIF
       </button>
    </>
   )
}

One takeaway here is that when the conversion request is successful, the user is redirect to the job page

Putting it all Together

// pages/index.tsx
import axios from "axios";
import { useRouter } from "next/router";
import React, { useMemo, useState } from "react";
import YouTube from "react-youtube";

const Home: React.FC = () => {
  const router = useRouter();

  const [youtubeUrl, setYoutubeUrl] = useState("");
  const [startTime, setStartTime] = useState("");
  const [endTime, setEndTime] = useState("");
  const [loading, setLoading] = useState(false);
  const [ytPlayer, setYtPlayer] = useState(null);

  const validYoutubeUrl = useMemo(() => {
    const youtubeUrlRegex = /^(?:https?:\/\/)?(?:www\.)?(?:youtu\.be\/|youtube\.com\/(?:embed\/|v\/|watch\?v=|watch\?.+&v=))((\w|-){11})(?:\S+)?$/;
    return youtubeUrl.match(youtubeUrlRegex);
  }, [youtubeUrl]);

  const ytVideoId = useMemo(() => {
    return youtubeUrl.split("v=")[1]?.slice(0, 11);
  }, [youtubeUrl]);

  const submitYoutubeVideo = async () => {
    setLoading(true);
    try {
      const response = await axios.post(
        `${process.env.NEXT_PUBLIC_BASE_URL}/api/v1/jobs`,
        {
          youtubeUrl,
          startTime: Number(startTime),
          endTime: Number(endTime),
        },
        {}
      );
      router.push(`/jobs/${response.data.id}`);
    } catch (err) {
      console.log(err);
      alert(err?.response?.data?.message || "Something went wrong");
    }
    setLoading(false);
  };
  return (
    <>
      {validYoutubeUrl ? (
        <>
          <h3>Preview</h3>
          <YouTube
            videoId={ytVideoId}
            opts={{
              playerVars: {
                start: Number(startTime),
                end: Number(endTime),
                autoplay: 0,
              },
            }}
            onReady={(e) => {
              setYtPlayer(e.target);
            }}
          />
        </>
      ) : (
        <h4>No Youtube Video Link Selected</h4>
      )}

      <input
        className={`input ${youtubeUrl === ""? "is-dark": validYoutubeUrl? "is-success": "is-danger"}`}
        type="text"
        placeholder="Youtube URL, eg: https://www.youtube.com/watch?v=I-QfPUz1es8"
        value={youtubeUrl}
        onChange={(e) => {
          setYoutubeUrl(e.target.value);
        }}
      />

      <input
        className="input is-dark"
        type="number"
        placeholder="Start Second, eg: 38"
        value={startTime}
        onChange={(e) => {
          setStartTime(e.target.value);
        }}
      />

      <input
        className="input is-dark"
        type="number"
        placeholder="End Second, eg: 72"
        value={endTime}
        onChange={(e) => {
          setEndTime(e.target.value);
        }}
      />

      <button
        className={`button is-black`}
        onClick={() => {
          if (ytPlayer)
            ytPlayer.loadVideoById({
              videoId: ytVideoId,
              startSeconds: Number(startTime),
              endSeconds: Number(endTime),
            });
        }}
      >
        Preview
      </button>

      <button
        className={`button is-black is-outlined ${loading ? "is-loading" : ""}`}
        onClick={submitYoutubeVideo}
      >
        Generate GIF
      </button>
    </>
  );
};

export default Home;

GIF Page

Polling the GIF Conversion Job

What we want to achieve here is periodically fetch the GIF Conversion Job data from the backend. This is known as polling.
To do this we are going to be using swr which is a data fetching library for React. It is not necessarily used for polling but it has a nice API that supports polling (refreshing data on an interval). Other data fetching libraries with similar capabilities exist most notably React Query. You can also perform polling with axios (using timeouts) however data fetching libraries like swr and React Query provide hooks for data fetching which improves the development experience as well as provide other capabilities such as caching.

First we have to provide the data fetching function

import axios from "axios";
import Job from "../../common/interfaces/Job.interface";

export default async function fetchJobById(jobId: string): Promise<Job> {
  try {
    const response = await axios.get(
      `${process.env.NEXT_PUBLIC_BASE_URL}/api/v1/jobs/${jobId}`
    );

    return response.data;
  } catch (err) {
    if (err.response?.status === 404) window.location.href = "/404";
    throw err;
  }
}

we can then use this with swr to poll our GIF conversion job

// pages/jobs/[id].tsx
import { useRouter } from "next/router";
import React from "react";
import useSWR from "swr";
import Job from "../../lib/common/interfaces/Job.interface";
import fetchJobById from "../../lib/requests/fetchers/jobById";

export default function JobPage() {
  const router = useRouter()
  const { jobId } = router.query
  const [jobDone, setJobDone] = React.useState(false);


  const { data: job, error: errorJob, isValidating: isValidatingJob } = useSWR(
    [`/api/jobs/${jobId}`, jobId],
    async (url, jobId) => await fetchJobById(jobId),
    {
      initialData: null,
      revalidateOnFocus: false,
  // job will be polled from the backend every 2 seconds until its status change to 'done'
      refreshInterval: jobDone ? 0 : 2000,
    }
  );

  React.useEffect(() => {
    if (job?.status === "done") setJobDone(true);
  }, [job]);

  const loadingJob = !job;

  return (
    <>
     {/* rendering logic */}
    </>
  );
}

Notice in that snippet that the refreshInterval is how often the data will be polled from the backend. we used a boolean state that will keep track of the job status and once it is done, we will stop polling the backend

Server Side Rendering

We can leverage Next's server side rendering to dynamically get the id from the url as well as initially fetch the job once before the page loads.
To do this we will use getServerSideProps()
See Next.js Docs for more info about this

// pages/jobs/[id].tsx
// ...other imports
import { InferGetServerSidePropsType } from "next";

export const getServerSideProps = async (context) => {
  const jobId = context.params.id;
  try {
    const initialJob: Job = await fetchJobById(jobId);
    return { props: { jobId, initialJob: initialJob } };
  } catch (err) {
    return { props: { jobId, initialJob: null } };
  }
};


export default function JobPage({
  jobId,
  initialJob,
}: InferGetServerSidePropsType<typeof getServerSideProps>)  {
  //...other code
  const { data: job, error: errorJob, isValidating: isValidatingJob } = useSWR(
    [`/api/jobs/${jobId}`, jobId],
    async (url, jobId) => await fetchJobById(jobId),
    {
      // use initialJob instead of null
      initialData: initialJob,
      revalidateOnFocus: false,
      refreshInterval: jobDone ? 0 : 2000,
    }
  );

  return (
    <>
     {/* rendering logic */}
    </>
  );
}

Notice that we used initialJob in the initialData property in swr options

Putting It All Together

// pages/jobs/[id].tsx
import { InferGetServerSidePropsType } from "next";
import React from "react";
import useSWR from "swr";
import Job from "../../lib/common/interfaces/Job.interface";
import fetchJobById from "../../lib/requests/fetchers/jobById";

export default function JobPage({
  jobId,
  initialJob,
}: InferGetServerSidePropsType<typeof getServerSideProps>) {
  const [jobDone, setJobDone] = React.useState(false);

  const { data: job, error: errorJob, isValidating: isValidatingJob } = useSWR(
    [`/api/jobs/${jobId}`, jobId],
    async (url, jobId) => await fetchJobById(jobId),
    {
      initialData: initialJob,
      revalidateOnFocus: false,
      refreshInterval: jobDone ? 0 : 2000,
    }
  );

  React.useEffect(() => {
    if (job?.status === "done") setJobDone(true);
  }, [job]);

  const loadingJob = !job;

  return (
    <>
      {loadingJob ? (
        <>
          <h4>Getting conversion status..</h4>
          <progress className="progress is-medium is-dark" max="100">
            45%
          </progress>
        </>
      ) : (
        <div className="content">
          {job.status === "error" ? (
            <h4 style={{ color: "#FF0000" }}>Conversion Failed</h4>
          ) : job.status === "done" ? (
            <>
              {!job.gifUrl ? (
                <h4 style={{ color: "#FF0000" }}>Conversion Failed</h4>
              ) : (
                <>
                  <h4>Gif</h4>
                  <img src={job.gifUrl}></img>
                  <h6>
                    GIF Url : <a href={job.gifUrl}>{job.gifUrl}</a>
                  </h6>
                  <h6>
                    Converted from :
                    <a href={job.youtubeUrl}>{job.youtubeUrl}</a>
                  </h6>
                </>
              )}
            </>
          ) : (
            <>
              <h4>Working..</h4>
              <h5>Conversion Status : {job.status}</h5>
              <progress className="progress is-medium is-dark" max="100">
                45%
              </progress>
            </>
          )}
        </div>
      )}
    </>
  );
}

export const getServerSideProps = async (context) => {
  const jobId = context.params.id;
  try {
    const initialJob: Job = await fetchJobById(jobId);
    return { props: { jobId, initialJob: initialJob } };
  } catch (err) {
    return { props: { jobId, initialJob: null } };
  }
};

This was the last part of our series! Hopefully you learned something new and remember that the full source code can be viewed on the github repository

Node Worker Service - Youtube GIF Maker Using Next.js, Node and RabbitMQ

Raggi — Sun, 14 Feb 2021 22:15:44 +0000

Hello everyone,
This Article is the third part of the series Youtube GIF Maker Using Next.js, Node and RabbitMQ.

In this article we will dive into building the worker service of our Youtube to GIF converter. This Article will contain some code snippets but the whole project can be accessed on github which contains the full source code. You can also view the app demo. The following topics will be covered here

Functionalities
Flow Chart
Implementation
- RabbitMQ Service
  - Consuming Tasks From the Queue
  - Message Acknowledgment
- Conversion Service
  - Downloading Youtube Video
  - Converting Video to GIF
  - Uploading GIF to Google Cloud Storage
  - Putting It All Together
Closing Thoughts

Functionalities

As you can see, the service worker is responsible for:

Consuming tasks from the task queue
Converting a part of a youtube video to a GIF
Uploading the GIF to a cloud storage
Updating the job gifUrl and status in database

Flow Chart

This flow chart will simplify how the service worker is works

Implementation

RabbitMQ Service

Consuming Tasks From the Queue

Just like the RabbitMQ Service from the backend server in the previous part of this series, the RabbitMQ Service in the service worker is similar except for one single function, startConsuming()

//rabbitmq.service.ts
import amqp, { Channel, Connection, ConsumeMessage } from 'amqplib';
import Container, { Service } from 'typedi';
import { Job } from '../entities/jobs.entity';
import ConversionService from './conversion.service';

@Service()
export default class RabbitMQService {
  private connection: Connection;
  private channel: Channel;
  private queueName = 'ytgif-jobs';
  constructor() {
    this.initializeService();
  }

  private async initializeService() {
    try {
      await this.initializeConnection();
      await this.initializeChannel();
      await this.initializeQueues();
      await this.startConsuming();
    } catch (err) {
      console.error(err);
    }
  }
  private async initializeConnection() {
    try {
      this.connection = await amqp.connect(process.env.NODE_ENV === 'production' ? process.env.RABBITMQ_PROD : process.env.RABBITMQ_DEV);
      console.info('Connected to RabbitMQ Server');
    } catch (err) {
      throw err;
    }
  }

  private async initializeChannel() {
    try {
      this.channel = await this.connection.createChannel();
      console.info('Created RabbitMQ Channel');
    } catch (err) {
      throw err;
    }
  }

  private async initializeQueues() {
    try {
      await this.channel.assertQueue(this.queueName, {
        durable: true,
      });
      console.info('Initialized RabbitMQ Queues');
    } catch (err) {
      throw err;
    }
  }

  public async startConsuming() {
    const conversionService = Container.get(ConversionService);
    this.channel.prefetch(1);
    console.info(' 🚀 Waiting for messages in %s. To exit press CTRL+C', this.queueName);
    this.channel.consume(
      this.queueName,
      async (msg: ConsumeMessage | null) => {
        if (msg) {
          const job: Job = JSON.parse(msg.content.toString());
          console.info(`Received new job 📩 `, job.id);
          try {
            await conversionService.beginConversion(
              job,
              () => {
                this.channel.ack(msg);
              },
              () => {
                this.channel.reject(msg, false);
              },
            );
          } catch (err) {
            console.error('Failed to process job', job.id, err);
          }
        }
      },
      {
        noAck: false,
      },
    );
  }
}

startConsuming() will consume a message from the queue, parse its JSON object and then delegate the conversion process to the ConversionService.
All the ConversionService needs to do the conversion process is the Job object as well as two callbacks used to either acknowledge or reject the message from the queue (Will be discussed below).

Also notice that in this example we use

this.channel.prefetch(1);

We will talk about this at the end of this part of the series and what it means

Message Acknowledgment

To remove a task from the queue (indicating that the service successfully processed the task either negatively or positively) we need to do manual acknowledgment.

This can be done in amqplib by using either

channel.ack(msg);

To indicate a positive message acknowledgement
or

// Second parameter specifies whether to re-queue the message or not
channel.reject(msg, false);

To indicate a negative message acknowledgement.
Notice that on error we do not re-queue the message back to the queue and we consider it as a 'failed conversion'. But this can be left up to the programmer to handle.

See more on RabbitMQ Message Acknowledgement

Conversion Service

This service contains the core logic of our service worker.

It exposes a function beginConversion() that is called from the RabbitMQ Service when consuming a message

public async beginConversion(job: Job, { onSuccess, onError }: { onSuccess: () => void; onError: () => void }) {
...
}

This function will perform all the steps necessary for the conversion, it will then call either onSuccess() or onError() depending on its success or failure.

These are the steps neccesary for converting a youtube video to a GIF:

Downloading Youtube Video
- The youtube video is downloaded locally
Converting downloaded video to GIF
- The video is converted into a GIF (only the selected range by start/end times is converted)
Uploading GIF to Google Cloud Storage
Updating the database
call onSuccess() or onError() accordingly

Lets start by downloading the youtube video locally

Downloading Youtube Video

to download the youtube video locally, we use the go-to package for that task, ytdl-core.

a function downloadVideo() is responsible for this, it takes the youtube video url/id and returns a ReadableStream that we can use to save the video file locally as well as its extension i.e: mp4, avi..etc

//conversion.service.ts
import { Readable } from 'stream';
import ytdl from 'ytdl-core';
import YoutubeDownload from '../common/interfaces/YoutubeDownload';


private async downloadVideo({ youtubeId, youtubeUrl }: YoutubeDownload): Promise<{ video: Readable ; formatExtension: string }> {
    const info = await ytdl.getInfo(youtubeId);
    const format: ytdl.videoFormat = info.formats[0];
    if (!format) throw new Error('No matching format found');
    const video = ytdl(youtubeUrl, {
      format,
    });
    return { video, formatExtension: format.container };
  }

public async beginConversion(job: Job, { onSuccess, onError }: { onSuccess: () => void; onError: () => void }) {
    try {
      console.info('Started Processing Job :', job.id);

      const { video, formatExtension } = await this.downloadVideo({
        youtubeId: job.youtubeId,
        youtubeUrl: job.youtubeUrl,
      });

      const srcFileName = `./src/media/temp.${formatExtension}`;

      video.on('progress', (chunkLength, downloaded, total) => {
        //... Logic for showing progress to the user..i.e progress bar
      });

      video.pipe(
        fs
          .createWriteStream(srcFileName)
          .on('open', () => {
            //Video download started
            console.log('Downloading Video');
          })
          .on('finish', async () => {
            //Video finished downloading locally in srcFileName
            console.info('Downloaded video for job ', job.id);
            //...Logic for converting the locally downloaded video to GIF
          })
          .on('error', async () => {
              //...handle failure logic
          }),
      );
    } catch (err) {
        //...handle failure logic
    }
  }

Converting Video to GIF

To convert local videos to GIFs we will use ffmpeg.wasm which is essentially a Webassembly port of FFmpeg. So you can think of this process as using FFmpeg inside node asynchronously to do the conversion. no spawning external processes, no dependent tools ...etc which is very powerful and simple.

//conversion.service.ts
import { createFFmpeg, fetchFile, FFmpeg } from '@ffmpeg/ffmpeg';
import GifConversion from '../common/interfaces/GifConversion';

//...somewhere in our code 
const ffmpeg = createFFmpeg({
        log: false,
        progress: p => {
          progressBar.update(Math.floor(p.ratio * 100));
        },
      });
await ffmpeg.load();

  //Converts a video range to GIF from srcFileName to destFileName 
private async convertToGIF({ startTime, endTime, srcFileName, destFileName, formatExtension }: GifConversion) {
    try {
      console.info('Converting Video to GIF');
      this.ffmpeg.FS('writeFile', `temp.${formatExtension}`, await fetchFile(srcFileName));
      await this.ffmpeg.run(
        '-i',
        `temp.${formatExtension}`,
        '-vcodec',
        'gif',
        '-ss',
        `${startTime}`,
        '-t',
        `${endTime - startTime}`,
        '-vf',
        'fps=10',
        `temp.gif`,
      );
      await fs.promises.writeFile(destFileName, this.ffmpeg.FS('readFile', 'temp.gif'));
      console.info('Converted video to gif');
    } catch (err) {
      throw err;
    }
  }

public async beginConversion(job: Job, { onSuccess, onError }: { onSuccess: () => void; onError: () => void }) {
    try {
      console.info('Started Processing Job :', job.id);

      const srcFileName = `./src/media/temp.${formatExtension}`;
      const destFileName = `./src/media/temp.gif`;

      //... Video download logic
      // GIF Conversion
      await this.convertToGIF({
        startTime: job.startTime,
        endTime: job.endTime,
        srcFileName,
        destFileName,
        formatExtension,
      });
    } catch (err) {
        //...handle failure logic
    }
  }

Uploading GIF to Google Cloud Storage

After the local video file is converted to a GIF, we can finally upload it to Google Cloud Storage.
First we will have a CloudStorageService that's responsible for just that!
in our case we use Google Cloud Storage.

import { Storage } from '@google-cloud/storage';
import * as _ from 'lodash';
import { Service } from 'typedi';

@Service()
class CloudStorageService {
  private storage;
  private BUCKET_NAME;
  constructor() {
    const privateKey = _.replace(process.env.GCS_PRIVATE_KEY, new RegExp('\\\\n', 'g'), '\n');
    this.BUCKET_NAME = 'yourbucketname';
    this.storage = new Storage({
      projectId: process.env.GCS_PROJECT_ID,
      credentials: {
        private_key: privateKey,
        client_email: process.env.GCS_CLIENT_EMAIL,
      },
    });
  }

  async uploadGif(gifImage: Buffer, uploadName: string) {
    try {
      const bucket = await this.storage.bucket(this.BUCKET_NAME);
      uploadName = `ytgif/${uploadName}`;
      const file = bucket.file(uploadName);
      await file.save(gifImage, {
        metadata: { contentType: 'image/gif' },
        public: true,
        validation: 'md5',
      });
      return `https://storage.googleapis.com/${this.BUCKET_NAME}/${uploadName}`;
    } catch (err) {
      throw new Error('Something went wrong while uploading image');
    }
  }
}

export default CloudStorageService;

we can now use it like that to upload the generated GIF

//conversion.service.ts
import Container from 'typedi';
import CloudStorageService from './cloudStorage.service';

private async uploadGifToCloudStorage(destFileName, uploadName): Promise<string> {
    try {
      console.info('Uploading gif to cloud storage');
      const gifImage = await fs.promises.readFile(destFileName);
      const cloudStorageInstance = Container.get(CloudStorageService);
      const gifUrl = await cloudStorageInstance.uploadGif(gifImage, `gifs/${uploadName}`);
      return gifUrl;
    } catch (err) {
      throw err;
    }
  }

public async beginConversion(job: Job, { onSuccess, onError }: { onSuccess: () => void; onError: () => void }) {
    try {
      const destFileName = `./src/media/temp.gif`;

      //... Video download logic
      //... Video conversion logic
      const gifUrl = await this.uploadGifToCloudStorage(destFileName, job.id);

    } catch (err) {
        //...handle failure logic
    }
  }

Handling success/failure

Handling success and failure is pretty simple. First, we have to update the job in the database

In case of success:
Set the job status to 'done' and update the gifUrl to the uploaded gif to Google Cloud Storage.

In case of failure:
Set the job status to 'error'

After that we will call onSuccess() or onError() which essentially will handle the positive/negative RabbitMQ message acknowledgment

public async beginConversion(job: Job, { onSuccess, onError }: { onSuccess: () => void; onError: () => void }) {
    try {
      const destFileName = `./src/media/temp.gif`;
      //... Video download logic
      //... Video conversion logic
      const gifUrl = await this.uploadGifToCloudStorage(destFileName, job.id);
      //Success scenario
      await this.jobService.updateJobById(job.id as any, { status: 'done', gifUrl });
      console.info(`Finished job ${job.id}, gif at ${gifUrl}`);
      onSuccess();
    } catch (err) {
        //Failure scenario
        console.error('Failed to process job', job.id);
        await this.jobService.updateJobById(job.id as any, { status: 'error' });
        onError();
    }
  }

Putting it all together

Putting it all together as well as adding a cli progress by with cli-progress the ConversionService looks like this

import Container, { Service } from 'typedi';
import JobsService from './jobs.service';
import ytdl from 'ytdl-core';
import { Readable } from 'stream';
import { Job } from '../entities/jobs.entity';
import { createFFmpeg, fetchFile, FFmpeg } from '@ffmpeg/ffmpeg';
import fs from 'fs';
import cliProgress from 'cli-progress';
import CloudStorageService from './cloudStorage.service';
import GifConversion from '../common/interfaces/GifConversion';
import YoutubeDownload from '../common/interfaces/YoutubeDownload';

const progressBar = new cliProgress.SingleBar({}, cliProgress.Presets.shades_classic);

@Service()
export default class ConversionService {
  private ffmpeg: FFmpeg = null;

  constructor(private jobService = new JobsService()) {}

  public async initializeService() {
    try {
      this.ffmpeg = createFFmpeg({
        log: false,
        progress: p => {
          progressBar.update(Math.floor(p.ratio * 100));
        },
      });
      await this.ffmpeg.load();
    } catch (err) {
      console.error(err);
    }
  }

  private async downloadVideo({ youtubeId, youtubeUrl }: YoutubeDownload): Promise<{ video: Readable; formatExtension: string }> {
    const info = await ytdl.getInfo(youtubeId);
    const format: ytdl.videoFormat = info.formats[0];
    if (!format) throw new Error('No matching format found');
    const video = ytdl(youtubeUrl, {
      format,
    });
    return { video, formatExtension: format.container };
  }

  private async convertToGIF({ startTime, endTime, srcFileName, destFileName, formatExtension }: GifConversion) {
    try {
      console.info('Converting Video to GIF');
      this.ffmpeg.FS('writeFile', `temp.${formatExtension}`, await fetchFile(srcFileName));
      progressBar.start(100, 0);
      await this.ffmpeg.run(
        '-i',
        `temp.${formatExtension}`,
        '-vcodec',
        'gif',
        '-ss',
        `${startTime}`,
        '-t',
        `${endTime - startTime}`,
        '-vf',
        'fps=10',
        `temp.gif`,
      );
      progressBar.stop();
      await fs.promises.writeFile(destFileName, this.ffmpeg.FS('readFile', 'temp.gif'));
      console.info('Converted video to gif');
    } catch (err) {
      throw err;
    }
  }

  private async uploadGifToCloudStorage(destFileName, uploadName): Promise<string> {
    try {
      console.info('Uploading gif to cloud storage');
      const gifImage = await fs.promises.readFile(destFileName);
      const cloudStorageInstance = Container.get(CloudStorageService);
      const gifUrl = await cloudStorageInstance.uploadGif(gifImage, `gifs/${uploadName}`);
      return gifUrl;
    } catch (err) {
      throw err;
    }
  }

  public async beginConversion(job: Job, { onSuccess, onError }: { onSuccess: () => void; onError: () => void }) {
    try {
      await this.jobService.updateJobById(job.id as any, { status: 'processing' });
      console.info('Started Processing Job :', job.id);

      const { video, formatExtension } = await this.downloadVideo({
        youtubeId: job.youtubeId,
        youtubeUrl: job.youtubeUrl,
      });

      const srcFileName = `./src/media/temp.${formatExtension}`;
      const destFileName = `./src/media/temp.gif`;

      video.on('progress', (chunkLength, downloaded, total) => {
        let percent: any = downloaded / total;
        percent = percent * 100;
        progressBar.update(percent);
      });

      video.pipe(
        fs
          .createWriteStream(srcFileName)
          .on('open', () => {
            console.log('Downloading Video');
            progressBar.start(100, 0);
          })
          .on('finish', async () => {
            progressBar.stop();
            console.info('Downloaded video for job ', job.id);

            await this.convertToGIF({
              startTime: job.startTime,
              endTime: job.endTime,
              srcFileName,
              destFileName,
              formatExtension,
            });

            const gifUrl = await this.uploadGifToCloudStorage(destFileName, job.id);

            await this.jobService.updateJobById(job.id as any, { status: 'done', gifUrl });
            console.info(`Finished job ${job.id}, gif at ${gifUrl}`);
            onSuccess();
          })
          .on('error', async () => {
            progressBar.stop();
            console.error('Failed to process job', job.id);
            await this.jobService.updateJobById(job.id as any, { status: 'error' });
            onError();
          }),
      );
    } catch (err) {
      await this.jobService.updateJobById(job.id as any, { status: 'error' });
      onError();
      throw err;
    }
  }
}

Closing Thoughts

Remember how we used channel.prefetch(1) when we started consuming from the queue

this.channel.prefetch(1);

What this does it makes sure that each queue consumer gets only on message at a time. This ensures that the load will be distributed evenly among our consumers and whenever a consumer is free it will be ready to process more tasks.
Read more about this from RabbitMQ Docs.

This also mean that if we want to scale our conversion process jobs/worker services we can add more replicas of this service.
Read more about this Competing Consumers

Thats it for our service worker! Now we can start digging into the client side of the app!.

Remember that the full source code can be viewed on the github repository

In the next part of the series we will see how we can implement The Next.js Client which will send gif conversion requests and view converted GIFs!.

Node Backend Server - Youtube GIF Maker Using Next.js, Node and RabbitMQ

Raggi — Sun, 14 Feb 2021 22:14:16 +0000

Hello everyone,
This Article is the second part of the series Youtube GIF Maker Using Next.js, Node and RabbitMQ.

In this article we will dive into building the backend server of our Youtube to GIF converter. This Article will contain some code snippets but the whole project can be accessed on github which contains the full source code as well as additional integration tests and swagger api docs. You can also view the app demo. The following topics will be covered here

Functionalities
Project Architecture
Implementation
- Database Schema
- Route Handling
- Controller
- Services
  - Job Service
  - RabbitMQ Service

Functionalities

As seen in the sequence diagram above, the backend server has 3 main functionalities which are:

Handling GIF conversion requests by creating a new job record in the database
Dispatching events to RabbitMQ indicating that a new conversion job has been created (task queueing)
Handling job fetching requests by querying the job by its id from the database and returning the appropriate response.

Project Architecture

Our express app architecture contains three main components

Route Handler
Controller
Service

Each one of them has certain functionalities that we will discuses in some detail as well as explain why its structured this way

Route Handler
- Responsible for routing paths to their route handlers. Typically these route handlers are composed of an array of handlers which we call the "Middleware Chain", the final handler in that chain is the Route Controller
- The Middleware Chain typically is responsible for doing "checks" on the incoming request as well as modifying the request object in some cases. In our case we will be doing validation using a custom validation middleware.
Controller
- Extracting data from the request as well as sanitizing this data if necessary
- Delegating the control to the relevant Service
- Handling responses
- Delegating errors to a custom error handling middleware
Service
- Has all the business logic
- Access data using the Data Access Layer (ORM/ODM)

Controllers should be dumb meaning they shouldn't have any details about the business logic, all they know is "which service can handle this request" , "what data this service needs" , "what the response should look like". This avoids having Fat Controllers

Implementation

Database Schema

In this project, we are using TypeORM which is a TypeScript ready ORM that supports many databases (we are going to be using MongoDB as mentioned in the first part of the series).

We are going to represent each GIF Conversion as a Job, which will be our only Collection.
The Job Collection in TypeORM looks like this



import { BaseEntity, Entity, ObjectID, Column, CreateDateColumn, UpdateDateColumn, ObjectIdColumn } from 'typeorm';

@Entity('jobs')
export class Job extends BaseEntity {
  @ObjectIdColumn()
  id: ObjectID;

  @Column({
    nullable: false,
  })
  youtubeUrl: string;

  @Column({
    nullable: false,
  })
  youtubeId: string;

  @Column({
    nullable: true,
  })
  gifUrl: string;

  @Column({
    nullable: false,
  })
  startTime: number;

  @Column({
    nullable: false,
  })
  endTime: number;

  @Column({
    type: 'enum',
    enum: ['pending', 'processing', 'done', 'error'],
  })
  status: 'pending' | 'processing' | 'done' | 'error';

  @Column()
  @CreateDateColumn()
  createdAt: Date;

  @Column()
  @UpdateDateColumn()
  updatedAt: Date;
}

The important thing to notice here is the field status, this essentially acts as an enum to indicate the current status of the GIF conversion. All the other fields are standard data needed to do the conversion job

Route Handling

As mentioned before , we are going to only have two routes.

Route for creating a new GIF conversion job
Route for fetching data about a conversion job from its id which will be used for polling later by the client side

This is how our route handler looks like



//routes.interface
import { Router } from 'express';

interface Route {
  path?: string;
  router: Router;
}

export default Route;



//jobs.route.ts
import { Router } from 'express';
import { CreateJobDto } from '../../common/dtos/createJob.dto';
import Route from '../../common/interfaces/routes.interface';
import JobsController from '../../controllers/jobs.controller';
import validationMiddleware from '../middlewares/validation.middleware';

class JobsRoute implements Route {
  public path = '/jobs';
  public router = Router();

  constructor(private jobsController = new JobsController()) {
    this.initializeRoutes();
  }

  private initializeRoutes() {
    this.router.get(`${this.path}/:id`, this.jobsController.getJobById);
    this.router.post(`${this.path}`, validationMiddleware(CreateJobDto, 'body'), this.jobsController.createJob);
  }
}

export default JobsRoute;

For validation we use a custom validation middleware that validates a DTO using class-validator and class-transformer



//createJob.dto
import { Expose } from 'class-transformer';
import { IsNotEmpty, IsNumber, IsString, Matches } from 'class-validator';
import { IsGreaterThan } from './validators/isGreaterThan';
import { MaximumDifference } from './validators/maximumDifference';

export class CreateJobDto {
  @IsNotEmpty()
  @IsString()
  @Matches(/^(?:https?:\/\/)?(?:www\.)?(?:youtu\.be\/|youtube\.com\/(?:embed\/|v\/|watch\?v=|watch\?.+&v=))((\w|-){11})(?:\S+)?$/, {
    message: 'Invalid youtube url',
  })
  @Expose()
  public youtubeUrl: string;

  @IsNotEmpty()
  @IsNumber()
  @Expose()
  public startTime: number;

  @IsNotEmpty()
  @IsNumber()
  @IsGreaterThan('startTime', {
    message: 'end time must be greater than start time',
  })
  @MaximumDifference('startTime', {
    message: 'maximum gif duration is 30 seconds',
  })
  @Expose()
  public endTime: number;
}

Notice that IsGreaterThan and MaximumDifference are custom class-validator validation decorators, essentially they look like this (more information on this can be found in class-validator docs)



//isGreaterThan.ts
import { registerDecorator, ValidationOptions, ValidationArguments } from 'class-validator';

export function IsGreaterThan(property: string, validationOptions?: ValidationOptions) {
  return function (object: Object, propertyName: string) {
    registerDecorator({
      name: 'isGreaterThan',
      target: object.constructor,
      propertyName: propertyName,
      constraints: [property],
      options: validationOptions,
      validator: {
        validate(value: any, args: ValidationArguments) {
          const [relatedPropertyName] = args.constraints;
          const relatedValue = (args.object as any)[relatedPropertyName];
          return typeof value === 'number' && typeof relatedValue === 'number' && value > relatedValue; 
        },
      },
    });
  };
}

MaximumDifference looks similar to this but its return looks like this instead



return typeof value === 'number' && typeof relatedValue === 'number' && value - relatedValue <= difference;

And now our validation middleware looks like this



validation.middleware.ts
import { plainToClass } from 'class-transformer';
import { validate, ValidationError } from 'class-validator';
import { RequestHandler } from 'express';

const validationMiddleware = (type: any, value: string | 'body' | 'query' | 'params' = 'body', skipMissingProperties = false): RequestHandler => {
  return (req, res, next) => {
    validate(plainToClass(type, req[value]), { skipMissingProperties }).then((errors: ValidationError[]) => {
      if (errors.length > 0) {
        const message = errors.map((error: ValidationError) => Object.values(error.constraints)).join(', ');
        res.status(400).send(message);
      } else {
        next();
      }
    });
  };
};

export default validationMiddleware;

Controller

Our controller looks pretty standard, the only takeaways are extracting the CreateJobDto object from the body using plainToClass from class-transformer with excludeExtraneousValues: true, which destructures only the exposed fields (having @Expose() decorator in the CreateJobDto class) more about this in class-transformer docs



//jobs.controllers.ts
import { plainToClass } from 'class-transformer';
import { NextFunction, Request, Response } from 'express';
import { CreateJobDto } from '../common/dtos/createJob.dto';
import { Job } from '../entities/jobs.entity';
import JobsService from '../services/jobs.service';

class JobsController {
  constructor(private jobService = new JobsService()) {}

  public createJob = async (req: Request, res: Response, next: NextFunction): Promise<void> => {
    try {
      const jobDto: CreateJobDto = plainToClass(CreateJobDto, req.body, { excludeExtraneousValues: true });
      const createdJob: Job = await this.jobService.createJob(jobDto);

      res.status(201).json(createdJob);
    } catch (error) {
      next(error);
    }
  };

  public getJobById = async (req: Request, res: Response, next: NextFunction): Promise<void> => {
    try {
      const jobId = req.params.id;
      const job: Job = await this.jobService.findJobById(jobId);

      const responseStatus = job.status === 'done' ? 200 : 202;
      res.status(responseStatus).json(job);
    } catch (error) {
      next(error);
    }
  };
}

export default JobsController;

Also its worth noting that the response status code of [GET] /job/{id} is 202 when the conversion job is still in processing. See Asynchronous Request-Response Pattern for more information on this

In case of an error, the error is passed to the error middleware, which is the last middleware in our express middleware chain and it looks like this:



//error.middleware.ts
import { NextFunction, Request, Response } from 'express';

import { isBoom, Boom } from '@hapi/boom';
import { logger } from '../../common/utils/logger';

function errorMiddleware(error: Boom | Error, req: Request, res: Response, next: NextFunction) {
  const statusCode: number = isBoom(error) ? error.output.statusCode : 500;
  const errorMessage: string = isBoom(error) ? error.message : 'Something went wrong';
  logger.error(`StatusCode : ${statusCode}, Message : ${error}`);

  return res.status(statusCode).send(errorMessage);
}
export default errorMiddleware;

You might notice we imported a package called Boom, we will talk about it later in the Services section

Services

Job Service

The JobService has all the business logic and access to the Data Access layer as well as communicating with the RabbitMQ Service to dispatch events to the queue



//jobs.service.ts
import * as Boom from '@hapi/boom';
import Container from 'typedi';
import { CreateJobDto } from '../common/dtos/createJob.dto';
import EventEmitter from '../common/utils/eventEmitter';
import { Job } from '../entities/jobs.entity';
import RabbitMQService from './rabbitmq.service';

class JobsService {
  private events = {
    JobCreated: 'JobCreated',
  };

  constructor() {
    this.intiializeEvents();
  }

  private intiializeEvents() {
    EventEmitter.on(this.events.JobCreated, (job: Job) => {
      const rabbitMQInstance = Container.get(RabbitMQService);
      rabbitMQInstance.sendToQueue(JSON.stringify(job));
    });
  }

  public async findJobById(jobId: string): Promise<Job> {
    const job: Job = await Job.findOne(jobId);
    if (!job) throw Boom.notFound();

    return job;
  }

  public async createJob(jobDto: CreateJobDto): Promise<Job> {
    const createdJob: Job = await Job.save({ ...jobDto, youtubeId: jobDto.youtubeUrl.split('v=')[1]?.slice(0, 11), status: 'pending' } as Job);
    EventEmitter.emit(this.events.JobCreated, createdJob);
    return createdJob;
  }
}

export default JobsService;

Right off the bat you can see two imports you may be unfamiliar with, we will quickly go through them and then explain each function in this class in details.

Boom
- Used to create http objects with a powerful, simple and friendly interface. you can see how easy it was throwing a 404 Not Found Error object
typedi
- TypeDI is a powerful dependency injection package that has many features. One of this features is having Singleton Services which is how we use it in our case.

Now lets go into more details into some functions in the class

intiializeEvents()

This function uses a global EventEmitter we use globally across the project to add a pub/sub layer. its as simple as



//eventEmitter.ts
import { EventEmitter } from 'events';
export default new EventEmitter();

and now we can start listening on events specifically an event we will emit later when creating a new job called 'JobCreated'



  // Defines all the events in our service
  private events = {
    JobCreated: 'JobCreated',
  };

  private intiializeEvents() {
    // Start listening for the event 'JobCreated'
    EventEmitter.on(this.events.JobCreated, (job: Job) => {
    // Get a singleton instance of our RabbitMQService
      const rabbitMQInstance = Container.get(RabbitMQService);
    // Dispatch an event containing the data of the created job
      rabbitMQInstance.sendToQueue(JSON.stringify(job));
    });
  }

See more information on Adding A Pub/Sub layer To Your Express Backend

createJob()

This function does exactly two things.

Creating a new job document in the database
Dispatching an event 'JobCreated' that a new job has been created, that way the event listener will handle the logic of dispatching that event to the RabbitMQ Service

RabbitMQ Service

This service is responsible for connecting to the RabbitMQ Server, creating a channel and initializing the queue which will be used for producing tasks (will be consumed by our service worker).

amqplib is used to as the client for our RabbitMQ Server



//rabbitmq.service.ts
import { Service } from 'typedi';
import amqp, { Channel, Connection } from 'amqplib';
import { logger } from '../common/utils/logger';

@Service()
export default class RabbitMQService {
  private connection: Connection;
  private channel: Channel;
  private queueName = 'ytgif-jobs';
  constructor() {
    this.initializeService();
  }

  private async initializeService() {
    try {
      await this.initializeConnection();
      await this.initializeChannel();
      await this.initializeQueues();
    } catch (err) {
      logger.error(err);
    }
  }
  private async initializeConnection() {
    try {
      this.connection = await amqp.connect(process.env.NODE_ENV === 'production' ? process.env.RABBITMQ_PROD : process.env.RABBITMQ_DEV);
      logger.info('Connected to RabbitMQ Server');
    } catch (err) {
      throw err;
    }
  }

  private async initializeChannel() {
    try {
      this.channel = await this.connection.createChannel();
      logger.info('Created RabbitMQ Channel');
    } catch (err) {
      throw err;
    }
  }

  private async initializeQueues() {
    try {
      await this.channel.assertQueue(this.queueName, {
        durable: true,
      });
      logger.info('Initialized RabbitMQ Queues');
    } catch (err) {
      throw err;
    }
  }

  public async sendToQueue(message: string) {
    this.channel.sendToQueue(this.queueName, Buffer.from(message), {
      persistent: true,
    });
    logger.info(`sent: ${message} to queue ${this.queueName}`);
  }
}

The code for bootstrapping the connection/channels/queues is pretty standard and you can find references to these functons on RabbitMQ Docs or anqplib docs. The one function that we will need to use from outside this class is sendToQueue() which is used to dispatch a message to our task queue as seen in the JobService by dispatching a stringified Job object.



 rabbitMQInstance.sendToQueue(JSON.stringify(job));

We now only need to initialize the RabbitMQ Service at the start of our app like this



import Container from 'typedi';

// Call initializeRabbitMQ() somewhere when starting the app
private initializeRabbitMQ() {
    Container.get(RabbitMqService);
  }

Now the job of our backend service is done and all that is left is for the node service worker to consume the task queue and do the actual GIF conversion.

Remember that the full source code can be viewed on the github repository

In the next part of the series we will see how we can implement a node service worker that will consume the task queue and do the actual Youtube to GIF conversion.

Getting Started - Youtube GIF Maker Using Next.js, Node and RabbitMQ

Raggi — Sun, 14 Feb 2021 22:13:47 +0000

Hello everyone,
This Article is the first part of the series Youtube GIF Maker Using Next.js, Node and RabbitMQ. You can find the full project on github and you can also view the app demo.

This article will cover

App Overview
- Basic Idea
- Functionalities
System Design
- Components
- Architecture
- Sequence Diagram

Overview

Idea

The basic idea of the app is to let users create GIFs from Youtube Videos. Users can do that by defining the url of the youtube video they want to make a GIF from as well as defining the start/end times of the GIF

App Functionalities

Create a GIF from a youtube video and the start/end times (range) from the video to make into a GIF
Preview the result before doing the actual conversion

System Design

System Components

Each component of the system is responsible for working in isolation and communicating with other components

React (Next.js) as the client side
- Communicates with the backend server to create new gif conversion requests as well as getting information of already created gif conversions
Node as a backend server
- Handles requests from the client as well as dispatching new conversion jobs to the message broker
Node as a service worker
- Will be responsible for executing/processing the conversion jobs
RabbitMQ as a message broker
- Will act as a Task Queue where the backend server will produce tasks to it and the service worker will consume tasks from it
MongoDB for datastore
- Will store information about the GIF conversion jobs
Google Cloud Storage for media storage
- Will be used to store the converted GIFs

Architecture

Sequence Diagram

This is a sequence diagram demonstrating the data flow including all the components mentioned above

In the next article in the series, we will start by diving deeper into the backend server

Asynchronous Request-Response Pattern

Raggi — Mon, 11 Jan 2021 13:31:01 +0000

The Asynchronous Request-Response Pattern lets you decouple backend services processing from a frontend client, where the backend processing needs to be asynchronous, while the frontend client needs a clear/near-immediate response.

Imagine a scenario of a web application that compresses video files. As you can imagine the process of compressing a video file can take time from seconds to minutes or even hours in some cases!. In that case, it is not really an option for the server to wait for this long running operation to finish before sending out the response.

To address this problem, backend engineers typically achieve backend asynchronous processing by adding a queue (achieved by using a message broker) that will isolate the backend processing (video compression in our case) from the request/response stages.

Problem

You can see how this approach can decouple the client which needs an almost immediate and clear response from the backend processing which needs to be done asynchronously which allows the client and the backend API to scale independently.
However, this approach adds another complexity, specifically when the clients needs to know when the job is done? (i.e video finished compressing) how can this be achieved in an asynchronous way?

Solution

This problem can be solved with 2 methods. HTTP Polling and WebSockets

HTTP Polling

HTTP Polling is where the client repeatedly sends a request to the server to get updated information, this information can be about a resource, a simple query or even updating or creating records in the database.
In our case, polling is used to check if the video has completed processing or not.

With this method the flow is as follows:

The client sends a request to the server to compress a video
The server adds the compression job to the queue (by emitting an event) and immediately responds to the client with a Job ID
The client can now start polling the server with the Job ID it received, for example, the server would provide an endpoint for checking the job status (GET /jobs/{id})
Now when the client polls the server, there are three possible scenarios, either the job is still processing, the job completed or the job failed. The server can response to either of these with different HTTP status codes. (202 Accepted, 200 OK, 400 Bad Request respectivly)

Note that instead of responding to the client with a Job ID, the server can also respond with a location url to poll the job status as well as the polling retrial interval (which can be an estimate of how long the processing might take to complete)

WebSockets

WebSockets can also be used to solve this problem, which is more suitable for when the client must receive the job status in real time.
The way it works is instead of polling for the job status, a WebSockets connection is established between the client and the server which keeps the client updated of the status of the job.

Adding A Pub/Sub layer To Your Express Backend

Raggi — Thu, 10 Dec 2020 15:36:48 +0000

Adding a Pub/Sub layer to your express backend can add an event-driven capability that makes handing certain operations more intuitive as well as provide better code separation.

Sometimes we might want to perform some actions or call third party services as a result of an event occurring in our app. For example sending a welcome email, a welcome sms or analytics data when a new user is registered which is very common in most apps these days.

Lets take the aforementioned example where we send email, sms and analytic data when a user registers. Traditionally this can be done by using imperative function calls as shown in the example below.

//auth.service.ts

import EmailService from './services/mail.service';
import SMSService from './services/sms.service';
import AnalyticsService from './services/analytics.service';
//...other imports

class AuthService {
  public async signup(userData): Promise<User> {
    const findUser: User = await User.findOne({ where: { email: userData.email } });
    if (findUser) throw new Error(`Email ${userData.email} already exists`);

    const hashedPassword = await bcrypt.hash(userData.password, 10);
    const createdUser: User = await User.save({ ...userData, password: hashedPassword });

    //Some actions
    AnalyticsService.addUserRecord({email:createdUser.email, number:createdUser.number});
    EmailService.sendWelcomeEmail(createdUser.email);
    //...Other user sign up actions
    SMSService.sendWelcomeSMS(createdUser.number);

    return createdUser;
  }
}

You can already see how this code will look like as we keep adding more actions, each action will add another imperative function call to a dependent service and the function will keep growing in size. You can also see that besides being hard to maintain, this approach violates the Single Responsibility Principle as well as has the potential for repetition across different events not only user registration.

Pub/Sub Layer

Adding a Pub/Sub layer can solve this problem by emitting an event (user registered with this email) and letting separate listeners handle the work.

We will utilize Node.js's Event Emitter to do that.

First we will create a shared Event Emitter as well as specify the set of events we need.

//eventEmitter.ts
import { EventEmitter } from 'events';

const Events = {
  USER_REGISTRATION = 'user-registered',
}
const eventEmitter = new EventEmitter();

export { eventEmitter, Events };

Note: Due to Node.jS Caching, this will always return the same instance of eventEmitter (Singleton)

Now we can modify our code to emit a "user registration event"

//auth.service.ts

import { eventEmitter, Events } from '../common/utils/eventEmitter';
//...other imports

class AuthService {
  public async signup(userData): Promise<User> {
    const findUser: User = await User.findOne({ where: { email: userData.email } });
    if (findUser) throw new Error(`Email ${userData.email} already exists`);

    const hashedPassword = await bcrypt.hash(userData.password, 10);
    const createdUser: User = await User.save({ ...userData, password: hashedPassword });

    //Emit User Registration Event
    eventEmitter.emit(Events.USER_REGISTRATION,{ email: userData.email, number: userData.number });

    return createdUser;
  }
}

Now Separate Services can listen on events and do their Job, For example the EmailService

//email.service.ts

import MailGunClient from '../common/clients/mailGun.client';
import EmailClient from '../common/interfaces/emailClient.interface';
import { eventEmitter, Events } from '../common/utils/eventEmitter';

class EmailService {
  constructor(private emailClient: EmailClient = new MailGunClient()) {
    this.initializeEventListeners();
  }

  private initializeEventListeners(): void {
    eventEmitter.on(Events.USER_REGISTRATION, ({ email }) => {
      this.emailClient.sendWelcomeEmail(email);
    });
  }
}

export default EmailService;

Now all that is left is to create an instance of your event listening services when bootstrapping your express app to initialize their listeners, something like calling this function when initializing your app

  private initializeServices() {
    new AnalyticsService();
    new EmailService();
    new SMSService();
  }

You can already see how adding more actions won't add any extra lines of code in the user registration function which provides code separation and embraces the event driven nature of Node.js.

DISJOINT SET UNION

Raggi — Sun, 15 Nov 2020 05:57:38 +0000

Disjoint Set Union (DSU)

Overview

The Disjoint-set data structure allows us to very quickly determine if two items are in the same set (or equivalently determine if two vertices are in the same connected component), and also to very quickly unite two sets (or equivalently combine two connected components into one connected component).

Interface

initSet(N)* Initializes an element and sets it to be its own set
unionSet(i,j)* Takes two elemements and combines them in one set
findSet(i)* Find the set an element belongs to
isSameSet(i,j) Find if two elements belong to the same set
numDisjointSets Find number of sets (connected componants)
sizeOfSet Find the size of a set an element belongs to

Data Structure

In its simplest form the DSU is an array of integers called parent where every ith item has its parent stored in the parent[i] or the array, These relationships create one, or more, virtual trees.

There you can see that 6 is under 7 in the tree, because parent[6] = 7. Also note that 2 and 3 have no parents, because parent[2] = 2 and parent[3] = 3.

Each tree is a disjoint set. If two elements are in the same tree, then they are in the same disjoint set. The root node (or the topmost node) of each tree is called the representative of the set. There is always a single unique representative of each set. As you can see, if i is the representative of a set, then parent[i] = i. If i is not the representative of his set, then it can be found by traveling up the tree until we find the representative.

Simple (Naive) implementation

We can already write a simple implementation of a DSU. It will be inefficient, but then we'll improve it with two techniques, resulting in an almost constant time of work.

The three main functions of the dsu are:

initSet(N) Initially each element belongs to its own set so we set it to be its own parent.
findSet(v)A simple recursive functions that climbs the parents of an element till it reaches the root of the set it belongs to.
unionSet(a,b)To combine two sets we firstly need to find if they belong to the same set (already unioned). If not then we assign them to the same set by setting the parents of j to be i.

vector < int > parent;

void initSet(int N)
{
    parent.assign(N, 0);
    for (int i = 0; i < N; i++)
        parent[i] = i;
}

int findSet(int i)
{
    if (i == parent[i])
        return i;
    return findSet(parent[i]);
}

void unionSet(int i, int j)
{
    i = findSet(i);
    j = findSet(j);
    if (i != j)
        parent[j] = i;
}

However, such implementation is considered to be inefficient as building a large tree will cause the depth of the different chains to be larger which will cause a worst case of O(n) for the findSet function

Efficient Implementation

To build a more efficient DSU there are two optimizations that can be done Path Compression and Combine by Rank.

Path Compression

Normally when using findSet to find a set of an element, the recursive function will traverse the whole parent chain of this element untill it reaches the representative or the effective parent of this set, This will cause the complexity of such operation to be of O(n) to avoid that we don't want to store the immediate ancestor for the element but store the top ancestor instead.

The new findSet() after applying the path compression concept should be as follows, notice that instead of returning the found parent, we save it before doing so.

The complexity after applying the path compression concept only would be O(log n) provided that unionSet() calls findSet() twice and does other O(1) operations that are ignored in the complexity calculation so the proof is concentrated on the O(log n) of findSet().

int findSet (int i) {
    if (i == parent[i])
        return i;
    return parent[i] = findSet (parent[i]);
}

Union by Rank

The previous improvement changed the findSet() operation to make the heights of the trees smaller. This improvement, called Union by Rank, changes the unionSet() operation to also make the heights of the trees smaller. The basic idea is that when joining two sets, the shorter set becomes the child of the longer one which will cause the height of the tree to be smaller.

This technique has two variants, one that ranks the trees based on tree size (number of vertices) and the other one is based on tree depth. In this example we will be implementing the tree depth union by rank variant.

To Acheive this a rank for each node is initially set to zero and then incremented every time a new node is added to it.

The complexity by using only the Union by Rank technique will effectivly be reduced to O(log n)

vector < int > parent;
vector < int > rank;

void initSet(int N)
{
    rank.assign(N, 0); //Rank is now initialized
    parent.assign(N, 0);
    for (int i = 0; i < N; i++)
        parent[i] = i;
}

void unionSet(int i, int j)
{
    i = findSet(i);
    j = findSet(j);
    if (i != j)
    {
        int x = findSet(i), y = findSet(j);
        if (rank[x] > rank[y])
            parent[y] = x;

        else
        {
            parent[x] = y;
            if (rank[x] == rank[y])
                rank[y]++; //Rank update
        }
    }
}

Complexity after improvements

The complexity of the DSU after adding either one of Path Compression or Union by Rank acheives an avarage of O(log n)

On combining both techniques together the complexity will achieve a near constant time O(1)

Putting it all together

The full class after applying Path Compression and Union by Rank. Also notice that three operations have been added

isSameSet(i,j) Find if two elements belong to the same set
numDisjointSets Find number of sets (connected componants)
sizeOfSet Find the size of a set an element belongs to

class UnionFind //OOP Style class
{ 
    private:
    vector <int>p, rank, setSize;
    int numSets;

    public:
    UnionFind(int N)
    {
        setSize.assign(N, 1);
        numSets = N;
        rank.assign(N, 0);
        p.assign(N, 0);
        for (int i = 0; i < N; i++)
            p[i] = i;
    }
    int findSet(int i) { return (p[i] == i) ? i : (p[i] = findSet(p[i])); }
    bool isSameSet(int i, int j) { return findSet(i) == findSet(j); }
    void unionSet(int i, int j)
    {
        if (!isSameSet(i, j))
        {
            numSets--;
            int x = findSet(i), y = findSet(j);
            if (rank[x] > rank[y])
            {
                p[y] = x;
                setSize[x] += setSize[y];
            }
            else
            {
                p[x] = y;
                setSize[y] += setSize[x];
                if (rank[x] == rank[y])
                    rank[y]++;
            }
        }
    }
    int numDisjointSets() { return numSets; }
    int sizeOfSet(int i) { return setSize[findSet(i)]; }
};</int>

Some Applications of DSU

Dynamically finding connected components in a graph
Searching for connected components in an image
Finding distances between a vertex and a representative of its set
Finding LCA in a tree

REST API Maturity - Towards The Glory of REST

Raggi — Sat, 14 Nov 2020 11:56:35 +0000

Is there a way to classify REST APIs based on their architectural maturity?

The Richardson Maturity Model developed by Leonard Richardson is a heuristic that is used to indicate how mature a web service is in regards to its REST Architectural style.

The model classifies APIs into 4 different levels (0-3)

Level 0 - The Swamp of POX
Level 1 - Resources
Level 2 - HTTP Verbs
Level 3 - Hypermedia Controls

The model identifies Level 3 as the Glory of REST

Level 0 - The Swamp of POX

In this level, HTTP is used only as a transport layer for remote interactions without using any inherent mechanisms of the web, meaning it only tunnels the request and nothing more. This style is usually associated with RPC (Remote Procedure Calls).
The acronym POX stand for Plain Old XML, this is because XML was used primarily as the payload of the request/response of the RPC however this idea can really be extended to any other formats, we will be using JSON for the examples.

Lets assume we have a social media app that lets users follow other users. Before following any specific user, we first want to show a list of all users. The app in this case will expose services at two separate endpoints.

An endpoint for listing of all users
An endpoint for following other users

First we will initiate a request to the first endpoint to list all users from Canada

 #Request
 POST /listUsers HTTP/1.1
 {
  "country" : "Canada"
 }

After the service receives the request, it will retrieve a list of all users living in Canada

 #Response
 HTTP/1.1 200 OK
 {
  [
    { "name" : "John Smith", "id" : 45 },
    { "name" : "Jane Doe", "id" : 22 }
  ]
 }

The next step now is to call the endpoint for following the desired user

 #Request
 POST /followUser HTTP/1.1
 {
  "id" : 22
 }

the response could be either a success or an error, in case of success it will look something like this:

 #Response
 HTTP/1.1 200 OK
 {
  "status" : "success"
 }

in case of an error it will look something like this:

 #Response
 HTTP/1.1 200 OK
 {
  "status" : "error",
  "message" : "already following"
 }

Notice that the request HTTP Method is always POST and the service HTTP status code is always 200 regardless of the state of the process.

Level 1 - Resources

In this level, the service starts working with resources denoted in the URI of the request. Note that resource identification is part of one of the main REST constraints which is having a Uniform Interface

The first request will be to the /users endpoint to get the list of all users, the response will be similar to the one before

 #Request
 POST /users HTTP/1.1
 {
  "country" : "Canada"
 }

 #Response
 HTTP/1.1 200 OK
 {
  [
    { "name" : "John Smith", "id" : 45 },
    { "name" : "Jane Doe", "id" : 22 }
  ]
 }

Now following the desired user by his id

 #Request
 POST /users/22 HTTP/1.1
 {}

The response could be either a success or an error

 #Response
 HTTP/1.1 200 OK
 {
  "status" : "success"
 }

 #Response
 HTTP/1.1 200 OK
 {
  "status" : "error",
  "message" : "already following"
 }

Notice that the HTTP Method as well as the HTTP status code remain unchanged just like Level 0 (POST and 200)

Level 2 - HTTP Verbs

Up until now, we have been strictly using HTTP POST, the idea was using HTTP only to tunnel the interactions through it and not utilizing any of its web mechanisms ( i.e: verbs and status codes ).

In this level we make sure to be as close as possible to the intent of the request in regards to the HTTP Verbs

The first request is now GET request to /users since we are logically reading or getting data, if we need to filter out based on a specific country we can now utilize query parameters

 #Request
 GET /users?country=canada HTTP/1.1

The response will be a list of users from Canada just like Level 0 and 1

 #Response
 HTTP/1.1 200 OK
 {
  [
    { "name" : "John Smith", "id" : 45 },
    { "name" : "Jane Doe", "id" : 22 }
  ]
 }

Now when following another user we use POST, you can think of it as if we are logically creating a new follow request

 #Request
 POST /users/22 HTTP/1.1
 {}

The response could be either a success or an error, in the case of success, it is denoted by the corresponding HTTP status code

 #Response
 HTTP/1.1 200 OK
 { }

In case of an error, a proper HTTP status code can be utilized as well

 #Response
 HTTP/1.1 409 CONFLICT
 {
  "message" : "already following"
 }

Level 3 - Hypermedia controls

Level 3 introduces using something you might have heard about here or there which is HATEOAS ( Hypermedia As The Engine Of Application State ). Essentially HATEOAS makes the API respond to requests with additional information and resource links that make exploring the API and accessing additional resources easier. Most of the time it can infer to the client as to what "actions" can be done next and where to access them if needed.

As before the request to get all users

 #Request
 GET /users?country=canada HTTP/1.1

Now we have the API respond with relevant hypermedia controls

 #Response
 HTTP/1.1 200 OK
 {
  [
    { 
      "name" : "John Smith", 
      "id" : 45 ,  
      "links" : [
       {
         "rel" : "rels/user/follow",
         "href" : "/users/45",
       }
      ]
    },
    { 
      "name" : "Jane Doe", 
      "id" : 22,  
      "links" : [
       {
         "rel" : "rels/user/follow",
         "href" : "/users/22",
       }
      ]
    },
  ]
 }

The rest of the requests/responses are pretty much the same as the previous level

 #Request
 POST /users/22 HTTP/1.1
 {}

success/failure

 #Response
 HTTP/1.1 200 OK
 { }

 #Response
 HTTP/1.1 409 CONFLICT
 {
  "message" : "already following"
 }

For many engineers, Level 2 seems to be enough and will put the API in a very solid state. You can imagine the complexity HATEOAS would introduce at both ends of the stack. However it has some benefits that shouldn't be just ignored.
HATEOAS provides the ability to seamlessly update the URI Schema of the APIs without worrying about or breaking clients as well as providing a self documenting API.

Closing Thoughts

Overall i think RMM in itself is not a model that defines how a truly RESTful API should look like, but it provides good guidelines that if followed can lead to a solid API Design head start.

Personally, i believe Level 3 is not worth the added complexity unless used in a very niche case where having a discoverable API is one of the key requirements of the project at hand.

I think introducing RMM especially to beginners in backend engineering is very important since it can make them think about good API Design and generally what makes an API RESTful by forcing them to atleast think about some of the constraints of REST even though it doesn't define it.