Expiration Service created:

- K8s & Docker config
- Added redis config
- Listener & publisher
- Added expiration-queue (bull)

Author: Shubhamjain2908

expiration/.dockerignore

@@ -0,0 +1 @@
+node_modules

expiration/Dockerfile

@@ -0,0 +1,8 @@
+FROM node:alpine
+
+WORKDIR /app
+COPY package.json .
+RUN npm install --only=prod
+COPY . .
+
+CMD [ "npm", "start" ]

expiration/package-lock.json

@@ -0,0 +1,33 @@
+{
+  "name": "expiration",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "start": "ts-node-dev --poll src/index.ts",
+    "test": "jest --watchAll --no-cache"
+  },
+  "jest": {
+    "preset": "ts-jest",
+    "testEnvironment": "node",
+    "setupFilesAfterEnv": [
+      "./src/test/setup.ts"
+    ]
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "@hn-hub/common": "^1.0.6",
+    "@types/bull": "^3.14.0",
+    "bull": "^3.15.0",
+    "node-nats-streaming": "^0.3.2",
+    "ts-node-dev": "^1.0.0-pre.49",
+    "typescript": "^3.9.5"
+  },
+  "devDependencies": {
+    "@types/jest": "^26.0.3",
+    "jest": "^26.1.0",
+    "ts-jest": "^26.1.1"
+  }
+}

expiration/package.json

@@ -0,0 +1,7 @@
+export const natsWrapper = {
+    client: {
+        publish: jest.fn().mockImplementation((subject: string, data: string, callback: () => void) => {
+            callback();
+        })
+    }
+};

expiration/src/__mocks__/nats-wrapper.ts

@@ -0,0 +1 @@
+export const queueGroupName = 'expiration-service';

expiration/src/events/listeners/queue-group-name.ts

@@ -0,0 +1,22 @@
+import { Listener, StoryCreatedEvent, Subjects } from '@hn-hub/common';
+import { Message } from 'node-nats-streaming';
+import { expirationQueue } from '../../queues/expiration-queue';
+import { queueGroupName } from './queue-group-name';
+
+export class StoryCreatedListener extends Listener<StoryCreatedEvent> {
+    subject: Subjects.StoryCreated = Subjects.StoryCreated;
+    queueGroupName = queueGroupName;
+
+    async onMessage(data: StoryCreatedEvent['data'], msg: Message) {
+        const delay = 60000;   // 10min delay (600000 ms)
+        console.log('Waiting this many milliseconds to process the Job: ', delay);
+        const stories = data.story.map(v => v.storyId);
+        await expirationQueue.add({
+            stories
+        }, {
+            delay
+        });
+
+        msg.ack();
+    }
+}

expiration/src/events/listeners/story-created-listener.ts

@@ -0,0 +1,5 @@
+import { ExpirationCompleteEvent, Publisher, Subjects } from '@hn-hub/common';
+
+export class ExpirationCompletePublisher extends Publisher<ExpirationCompleteEvent> {
+    subject: Subjects.ExpirationComplete = Subjects.ExpirationComplete;
+}

expiration/src/events/publisher/expiration-complete-publisher.ts

@@ -0,0 +1,40 @@
+
+import { StoryCreatedListener } from './events/listeners/story-created-listener';
+import { natsWrapper } from './nats-wrapper';
+
+const start = async () => {
+    console.log('Expiration service starting...');
+
+    if (!process.env.NATS_CLIENT_ID) {
+        throw new Error('NATS_CLIENT_ID must be defined');
+    }
+    if (!process.env.NATS_URL) {
+        throw new Error('NATS_URL must be defined');
+    }
+    if (!process.env.NATS_CLUSTER_ID) {
+        throw new Error('NATS_CLUSTER_ID must be defined');
+    }
+
+    try {
+        await natsWrapper.connect(
+            process.env.NATS_CLUSTER_ID,
+            process.env.NATS_CLIENT_ID,
+            process.env.NATS_URL
+        );
+
+        natsWrapper.client.on('close', () => {
+            console.log('Expiration Service : Nats connection closed!');
+            process.exit();
+        });
+
+        process.on('SIGINT', () => natsWrapper.client.close());
+        process.on('SIGTERM', () => natsWrapper.client.close());
+
+        new StoryCreatedListener(natsWrapper.client).listen();
+    } catch (err) {
+        console.error('Startup error => ', err);
+    }
+
+};
+
+start();

expiration/src/index.ts

@@ -0,0 +1,28 @@
+import nats, { Stan } from 'node-nats-streaming';
+
+class NatsWrapper {
+    private _client?: Stan;
+
+    get client() {
+        if (!this._client) {
+            throw new Error('Cannot access NATS client before connecting');
+        }
+        return this._client;
+    }
+
+    connect(clusterId: string, clientId: string, url: string) {
+        this._client = nats.connect(clusterId, clientId, { url });
+
+        return new Promise((resolve, reject) => {
+            this.client.on('connect', () => {
+                console.log('Connected to NATS');
+                resolve();
+            });
+            this.client.on('error', (err) => {
+                reject(err);
+            });
+        });
+    }
+}
+
+export const natsWrapper = new NatsWrapper();