feat(posts): initial live feed implementation (broadcasting)

docker-compose.yml

@@ -21,6 +21,21 @@ services:
         depends_on:
           - esi-nginx
 
+    zkill-nginx:
+        image: nginx:alpine
+        volumes:
+          - ./infra/zkill-cache/nginx-zkill.conf:/etc/nginx/nginx.conf:ro
+    zkill-varnish:
+        image: million12/varnish
+        environment:
+          - VCL_CONFIG=/varnish-zkill.vcl
+        volumes:
+          - ./infra/zkill-cache/varnish-zkill.vcl:/varnish-zkill.vcl:ro
+        links:
+          - zkill-nginx
+        depends_on:
+          - zkill-nginx
+
     api-image:
         build: .
         image: evebook/api
@@ -39,9 +54,11 @@ services:
         links:
           - postgres
           - esi-varnish
+          - zkill-varnish
         depends_on:
           - postgres
           - esi-varnish
+          - zkill-varnish
 
     updater:
         command: yarn run start:updater

infra/esi-cache/varnish-esi.vcl

@@ -3,6 +3,6 @@ vcl 4.0;
 # On Kuberentes is "localhost"
 # In Docker-Compose is "esi-nginx"
 backend default {
-    .host = "localhost";
+    .host = "esi-nginx";
     .port = "8080";
 }

infra/zkill-cache/nginx-zkill.conf

@@ -0,0 +1,21 @@
+
+worker_processes  1;
+
+events {
+    worker_connections  1024;
+}
+
+http {
+    include       mime.types;
+    default_type  application/octet-stream;
+
+    keepalive_timeout 65;
+
+    server {
+        listen 8080;
+
+        location / {
+            proxy_pass https://zkillboard.com;
+        }
+    }
+}

infra/zkill-cache/varnish-zkill.vcl

@@ -0,0 +1,8 @@
+vcl 4.0;
+
+# On Kuberentes is "localhost"
+# In Docker-Compose is "zkill-nginx"
+backend default {
+    .host = "zkill-nginx";
+    .port = "8080";
+}

src/modules/comment/comment.module.ts

@@ -11,6 +11,7 @@ import { CommandBus, CQRSModule, EventBus } from '@nestjs/cqrs';
 import { ModuleRef } from '@nestjs/core';
 import { eventHandlers } from './events/handlers';
 import { commandHandlers } from './commands/handlers';
+import { WebsocketModule } from '../websocket/websocket.module';
 
 @Module({
   imports: [
@@ -21,6 +22,7 @@ import { commandHandlers } from './commands/handlers';
     AllianceModule,
     CharacterModule,
     PostModule,
+    WebsocketModule,
   ],
   controllers: [
     CommentController,

src/modules/comment/events/handlers/create.handler.ts

@@ -6,13 +6,16 @@ import { CreateCommentEvent } from '../create.event';
 import { PostService } from '../../../post/post.service';
 import { Character } from '../../../character/character.entity';
 import { NOTIFICATION_TYPE } from '../../../notification/notification.constants';
+import { WebsocketGateway } from '../../../websocket/websocket.gateway';
+import { DComment } from '../../comment.dto';
 
 @EventsHandler(CreateCommentEvent)
 export class CreateCommentEventHandler implements IEventHandler<CreateCommentEvent> {
 
   constructor(
     private commandBus: CommandBus,
     private postService: PostService,
+    private websocketGateway: WebsocketGateway,
   ) {
   }
 
@@ -41,5 +44,10 @@ export class CreateCommentEventHandler implements IEventHandler<CreateCommentEve
       await this.commandBus.execute(new CreateNotificationCommand(notification));
     }
 
+    // Send comment to subscribers
+    this.websocketGateway.sendEventToPostCommentSub<DComment>(
+      event.comment.post,
+      new DComment(event.comment),
+    );
   }
 }

src/modules/core/external/zkillboard/zkillboard.service.ts

@@ -9,7 +9,7 @@ import {
 @Injectable()
 export class ZKillboardService {
 
-  private baseUrl = 'https://zkillboard.com/api/';
+  private baseUrl = `${process.env.ZKILLBOARD_ENDPOINT}/api/`;
   private userAgent = `eve-book/${process.env.npm_package_version} https://github.com/evebook/api`;
   private client: AxiosInstance;
 

src/modules/core/logger/logger.module.ts

@@ -1,18 +1,23 @@
 import { Global, Module } from '@nestjs/common';
 import { LoggerService } from './logger.service';
 import { loggerProviders } from './logger.providers';
-import { LoggerExceptionInterceptor } from './loggerException.interceptor';
+import {
+  HttpLoggerExceptionInterceptor,
+  WsLoggerExceptionInterceptor,
+} from './loggerException.interceptor';
 
 @Global()
 @Module({
   providers: [
     ...loggerProviders,
     LoggerService,
-    LoggerExceptionInterceptor,
+    HttpLoggerExceptionInterceptor,
+    WsLoggerExceptionInterceptor,
   ],
   exports: [
     LoggerService,
-    LoggerExceptionInterceptor,
+    HttpLoggerExceptionInterceptor,
+    WsLoggerExceptionInterceptor,
   ],
 })
 export class LoggerModule {

src/modules/core/logger/loggerException.interceptor.ts

@@ -7,10 +7,9 @@ import {
 import { Observable } from 'rxjs/Observable';
 import 'rxjs/add/operator/do';
 import { LoggerService } from './logger.service';
-import { Request } from 'express';
 
 @Injectable()
-export class LoggerExceptionInterceptor implements NestInterceptor {
+export class HttpLoggerExceptionInterceptor implements NestInterceptor {
 
   constructor(private loggerService: LoggerService) {
   }
@@ -45,3 +44,21 @@ export class LoggerExceptionInterceptor implements NestInterceptor {
     });
   }
 }
+
+@Injectable()
+export class WsLoggerExceptionInterceptor implements NestInterceptor {
+
+
+  intercept(
+    context: ExecutionContext,
+    call$: Observable<any>,
+  ): Observable<any> {
+    const client = context.switchToWs().getClient();
+
+    // first param would be for events, second is for errors
+    return call$.do(null, (exception) => {
+      console.log(exception.message);
+      console.error(JSON.parse(JSON.stringify(exception)));
+    });
+  }
+}

src/modules/notification/notification.dto.ts

@@ -1,5 +1,7 @@
 import {
-  NOTIFICATION_TYPE, WS_NOTIFICATION_EVENT, WS_NOTIFICATION_SEEN_EVENT,
+  NOTIFICATION_TYPE,
+  WS_NOTIFICATION_EVENT,
+  WS_NOTIFICATION_SEEN_EVENT,
 } from './notification.constants';
 import { Notification } from './notification.entity';
 import { WsResponse } from '@nestjs/websockets';

src/modules/post/events/handlers/create.handler.ts

@@ -0,0 +1,77 @@
+import { EventsHandler, IEventHandler } from '@nestjs/cqrs';
+import {
+  CharacterPostedOnAllianceWallEvent,
+  CharacterPostedOnCharacterWallEvent, CharacterPostedOnCorporationWallEvent, CharacterPostedEvent,
+} from '../create.character.event';
+import {
+  CorporationPostedOnCharacterWallEvent,
+  CorporationPostedOnCorporationWallEvent,
+  CorporationPostedOnAllianceWallEvent,
+  CorporationPostedEvent,
+} from '../create.corporation.event';
+import {
+  AlliancePostedOnCharacterWallEvent,
+  AlliancePostedOnCorporationWallEvent,
+  AlliancePostedOnAllianceWallEvent,
+  AlliancePostedEvent,
+} from '../create.alliance.event';
+import { PostedEvent } from '../create.event';
+import { WebsocketGateway } from '../../../websocket/websocket.gateway';
+import { Character } from '../../../character/character.entity';
+import { Corporation } from '../../../corporation/corporation.entity';
+import { Alliance } from '../../../alliance/alliance.entity';
+import { DPost } from '../../post.dto';
+
+@EventsHandler(
+  CharacterPostedEvent,
+  CharacterPostedOnCharacterWallEvent,
+  CharacterPostedOnCorporationWallEvent,
+  CharacterPostedOnAllianceWallEvent,
+  CorporationPostedEvent,
+  CorporationPostedOnCharacterWallEvent,
+  CorporationPostedOnCorporationWallEvent,
+  CorporationPostedOnAllianceWallEvent,
+  AlliancePostedEvent,
+  AlliancePostedOnCharacterWallEvent,
+  AlliancePostedOnCorporationWallEvent,
+  AlliancePostedOnAllianceWallEvent,
+)
+export class AnyonePostedEventHandler implements IEventHandler<PostedEvent> {
+
+  constructor(
+    private websocketGateway: WebsocketGateway,
+  ) {
+  }
+
+  async handle(event: PostedEvent) {
+    const post = new DPost(event.post);
+
+    // Send event for hashtags
+    for (const hashtag of event.post.hashtags) {
+      await this.websocketGateway.sendEventToHashtagWallSub<DPost>(hashtag, post);
+    }
+
+    // First check if it was posted on wall
+    // else check who posted it
+    const sendTo = event.post.characterWall
+      || event.post.corporationWall
+      || event.post.allianceWall
+      || event.post.character
+      || event.post.corporation
+      || event.post.alliance;
+
+    // and send it to correct wall
+    if (sendTo instanceof Character) {
+      await this.websocketGateway.sendEventToCharacterWallSub<DPost>(sendTo, post);
+    }
+    if (sendTo instanceof Corporation) {
+      await this.websocketGateway.sendEventToCorporationWallSub<DPost>(sendTo, post);
+    }
+    if (sendTo instanceof Alliance) {
+      await this.websocketGateway.sendEventToAllianceWallSub<DPost>(sendTo, post);
+    }
+
+    // Also send it to latest
+    await this.websocketGateway.sendEventToLatestWallSub<DPost>(post);
+  }
+}

src/modules/post/events/handlers/index.ts

@@ -1,5 +1,7 @@
 import { CharacterPostedOnWallEventHandler } from './create.character.wall.handler';
+import { AnyonePostedEventHandler } from './create.handler';
 
 export const eventHandlers = [
   CharacterPostedOnWallEventHandler,
+  AnyonePostedEventHandler,
 ];

src/modules/post/post.module.ts

@@ -16,6 +16,7 @@ import { ModuleRef } from '@nestjs/core';
 import { NotificationModule } from '../notification/notification.module';
 import { GooglePubSubModule } from '../core/googlePubSub/googlePubSub.module';
 import { GooglePubSub } from '../core/googlePubSub/googlePubSub';
+import { WebsocketModule } from '../websocket/websocket.module';
 
 @Module({
   imports: [
@@ -31,6 +32,7 @@ import { GooglePubSub } from '../core/googlePubSub/googlePubSub';
     HashtagModule,
     UniverseLocationModule,
     NotificationModule,
+    WebsocketModule,
   ],
   controllers: [
     PostController,

src/modules/websocket/websocket.constants.ts

@@ -1 +1,39 @@
+export const MAX_ROOMS_JOINED = 50;
+
 export const WS_EVENT_AUTHENTICATION = 'WS_EVENT_AUTHENTICATION';
+export const WS_EVENT_SUBSCRIPTION = 'WS_EVENT_SUBSCRIPTION';
+
+export const WS_NEW_SUBSCRIPTION_EVENT = 'WS_NEW_SUBSCRIPTION_EVENT';
+
+// Subscribe (Initial requests)
+export enum WS_SUBSCRIBE_EVENTS {
+    TO_LATEST_WALL = 'WS_EVENT_SUBSCRIBE_TO_LATEST_WALL',
+    TO_HASHTAG_WALL = 'WS_EVENT_SUBSCRIBE_TO_HASHTAG_WALL',
+    TO_CHARACTER_WALL = 'WS_EVENT_SUBSCRIBE_TO_CHARACTER_WALL',
+    TO_CORPORATION_WALL = 'WS_EVENT_SUBSCRIBE_TO_CORPORATION_WALL',
+    TO_ALLIANCE_WALL = 'WS_EVENT_SUBSCRIBE_TO_ALLIANCE_WALL',
+    TO_POST_COMMENTS = 'WS_EVENT_SUBSCRIBE_TO_POST_COMMENTS',
+}
+
+export enum WS_SUBSCRIPTIONS {
+    TO_LATEST_WALL = 'WS_EVENT_SUBSCRPTION_TO_LATEST_WALL',
+    TO_HASHTAG_WALL = 'WS_EVENT_SUBSCRPTION_TO_HASHTAG_WALL',
+    TO_CHARACTER_WALL = 'WS_EVENT_SUBSCRPTION_TO_CHARACTER_WALL',
+    TO_CORPORATION_WALL = 'WS_EVENT_SUBSCRPTION_TO_CORPORATION_WALL',
+    TO_ALLIANCE_WALL = 'WS_EVENT_SUBSCRPTION_TO_ALLIANCE_WALL',
+    TO_POST_COMMENTS = 'WS_EVENT_SUBSCRPTION_TO_POST_COMMENTS',
+}
+
+export const WS_EVENT_UNSUBSCRIBE_TO_LATEST_WALL = 'WS_EVENT_UNSUBSCRIBE_TO_LATEST_WALL';
+export const WS_EVENT_UNSUBSCRIBE_TO_HASHTAG_WALL = 'WS_EVENT_UNSUBSCRIBE_TO_HASHTAG_WALL';
+export const WS_EVENT_UNSUBSCRIBE_TO_CHARACTER_WALL = 'WS_EVENT_UNSUBSCRIBE_TO_CHARACTER_WALL';
+export const WS_EVENT_UNSUBSCRIBE_TO_CORPORATION_WALL = 'WS_EVENT_UNSUBSCRIBE_TO_CORPORATION_WALL';
+export const WS_EVENT_UNSUBSCRIBE_TO_ALLIANCE_WALL = 'WS_EVENT_UNSUBSCRIBE_TO_ALLIANCE_WALL';
+export const WS_EVENT_UNSUBSCRIBE_TO_POST_COMMENTS = 'WS_EVENT_UNSUBSCRIBE_TO_POST_COMMENTS';
+
+export const getRoomForLatestWall = () => 'wall:latest';
+export const getRoomForHashtagWall = (hashtag: string) => `wall:hashtag:${hashtag}`;
+export const getRoomForCharacterWall = (characterId: string) => `wall:character:${characterId}`;
+export const getRoomForCorporationWall = (corpId: string) => `wall:corporation:${corpId}`;
+export const getRoomForAllianceWall = (allianceId: string) => `wall:alliance:${allianceId}`;
+export const getRoomForPostComments = (postId: string) => `post:comments:${postId}`;

src/modules/websocket/websocket.dto.ts

@@ -1,4 +1,9 @@
-import { WS_EVENT_AUTHENTICATION } from './websocket.constants';
+import {
+  WS_EVENT_AUTHENTICATION,
+  WS_EVENT_SUBSCRIPTION,
+  WS_SUBSCRIPTIONS,
+  WS_NEW_SUBSCRIPTION_EVENT,
+} from './websocket.constants';
 import { WsResponse } from '@nestjs/websockets';
 
 export class DWsAuthentication implements WsResponse<{ success: boolean }> {
@@ -9,3 +14,29 @@ export class DWsAuthentication implements WsResponse<{ success: boolean }> {
     this.data = { success };
   }
 }
+
+export class DWsSubscription implements WsResponse<{ success: boolean, message?: string }> {
+  event = WS_EVENT_SUBSCRIPTION;
+  data: { success: boolean, message?: string };
+
+  constructor(success: boolean, message?: string) {
+    this.data = { success, message };
+  }
+}
+
+export class DWsNewSubscriptionEvent<T>
+  implements WsResponse<{subscription: WS_SUBSCRIPTIONS, payload: T}> {
+
+  event = WS_NEW_SUBSCRIPTION_EVENT;
+  data: {
+    subscription: WS_SUBSCRIPTIONS;
+    payload: T;
+  };
+
+  constructor(payload: T, subscription: WS_SUBSCRIPTIONS) {
+    this.data = {
+      subscription,
+      payload,
+    };
+  }
+}