neo4j start

SCHEMA.md

@@ -1,13 +1,21 @@
-(Chat { id, title, type })
-(User { id })
-(Message { id, text })
+(TGChat { id, title, type })
+(TGUser { id })
+(TGMessage { id, text })
 
-(User)-[:CREATED_MESSAGE { timestamp }]->(Message)
-(Message)-[:UPDATED_TO { timestamp }]->(Message)
-(Message)-[:MENTIONED]->(User)
-(User)-[:REACTED_TO { timestamp, old_reaction, new_reaction }]->(Message)
-(Message)-[:REPLY_TO]->(Message)
+(TGUser)-[:CREATED_MESSAGE { timestamp }]->(TGMessage)
+(TGMessage)-[:MESSAGE_EDITED { timestamp }]->(TGMessage)
+(TGMessage)-[:MENTIONED]->(User)
+(TGUser)-[:REACTED_TO { timestamp, old_reaction, new_reaction }]->(TGMessage)
+(TGMessage)-[:REPLIED]->(TGMessage)
 
 
-(User)-[:JOINED { timestamp }]->(Chat)
-(User)-[:LEFT { timestamp }]->(Chat)
+(TGUser)-[:JOINED { timestamp }]->(TGChat)
+(TGUser)-[:LEFT { timestamp }]->(TGChat)
+
+## Actions
+
+### Message
+
+1. Create or update chat
+2. Create or update user
+3. Create message

apps/bot/src/main.ts

@@ -1,8 +1,13 @@
 import { NestFactory } from '@nestjs/core';
 import { BotModule } from './bot.module';
+import { RmqService, Queues } from '@app/common';
 
 async function bootstrap() {
-  const app = await NestFactory.createMicroservice(BotModule);
-  app.listen();
+  // const app = await NestFactory.createMicroservice(BotModule);
+  const app = await NestFactory.create(BotModule);
+  const rmqService = app.get<RmqService>(RmqService);
+  app.connectMicroservice(rmqService.getOptions(Queues.Bot));
+  await app.startAllMicroservices();
+  app.listen(3000);
 }
 bootstrap();

apps/event-store/src/event-store.module.ts

@@ -2,7 +2,14 @@ import { Module } from '@nestjs/common';
 import { EventStoreController } from './event-store.controller';
 import { EventStoreService } from './event-store.service';
 import { ConfigModule, ConfigService } from '@nestjs/config';
-import { schemaConfig, rmqConfig, RmqModule, mongoConfig } from '@app/common';
+import {
+  schemaConfig,
+  rmqConfig,
+  RmqModule,
+  mongoConfig,
+  Queues,
+  Services,
+} from '@app/common';
 import { MongooseModule } from '@nestjs/mongoose';
 import { EventSchema, Event } from './schemas/event.schema';
 
@@ -22,6 +29,10 @@ import { EventSchema, Event } from './schemas/event.schema';
       inject: [ConfigService],
     }),
     MongooseModule.forFeature([{ name: Event.name, schema: EventSchema }]),
+    RmqModule.register({
+      name: Services.GraphStore,
+      queue: Queues.GraphStore,
+    }),
   ],
   controllers: [EventStoreController],
   providers: [EventStoreService],

apps/event-store/src/event-store.service.ts

@@ -1,13 +1,16 @@
-import { Injectable } from '@nestjs/common';
+import { Inject, Injectable } from '@nestjs/common';
 import { InjectConnection, InjectModel } from '@nestjs/mongoose';
 import { Connection, Model } from 'mongoose';
 import { Event, EventSchema } from './schemas/event.schema';
+import { Services } from '@app/common';
+import { ClientProxy } from '@nestjs/microservices';
 
 @Injectable()
 export class EventStoreService {
   constructor(
     @InjectModel(Event.name) private readonly eventModel: Model<Event>,
     @InjectConnection() private readonly connection: Connection,
+    @Inject(Services.GraphStore) private client: ClientProxy,
   ) {}
 
   async createEvent(
@@ -30,6 +33,8 @@ export class EventStoreService {
         event_type,
         event_data,
       });
+    console.log('Saved event to db.');
+    this.client.emit(event_type, createdEvent);
     return createdEvent;
   }
 }

apps/event-store/src/schemas/event.schema.ts

@@ -6,7 +6,7 @@ export type EventDocument = HydratedDocument<Event>;
 @Schema({ strict: false }) // strict: false allows storing unstructured data
 export class Event {
   @Prop({ required: true })
-  timestamp: Date;
+  timestamp: number;
 
   @Prop({ required: true })
   event_type: string;

apps/graph-store/src/cyphers/chat.cyphers.ts

@@ -0,0 +1,5 @@
+export const CREATE_OR_UPDATE_CHAT = `
+    MERGE(chat:TGChat { id: $chatId })
+    ON CREATE SET chat.title = $chatTitle, chat.type = $chatType, chat.timestamp = $timestamp
+    ON MATCH SET chat.title = $chatTitle, chat.type = $chatType, chat.timestamp = $timestamp
+  `;

apps/graph-store/src/cyphers/helpers/node.helper.spec.ts

@@ -0,0 +1,50 @@
+import { NodeHelper } from './node.helper';
+
+describe('NodeHelper', () => {
+  test('Basic Object', () => {
+    const user = {
+      id: 1,
+      is_bot: false,
+      first_name: 'John',
+      username: 'johndoe',
+    };
+
+    const nodeHelper = new NodeHelper();
+    const { query, params } = nodeHelper.createNode('user', 'TGUser', user);
+    expect(query).toEqual(`
+    MERGE (user:TGUser {id: $user_id})
+    ON CREATE SET user.is_bot = $user_is_bot, SET user.first_name = $user_first_name, SET user.username = $user_username
+    ON MATCH SET user.is_bot = $user_is_bot, SET user.first_name = $user_first_name, SET user.username = $user_username
+    `);
+    expect(params).toEqual({
+      user_id: 1,
+      user_is_bot: false,
+      user_first_name: 'John',
+      user_username: 'johndoe',
+    });
+  });
+
+  test('Id Only', () => {
+    const user = { id: 3 };
+    const nodeHelper = new NodeHelper();
+    const { query, params } = nodeHelper.createNode('user', 'TGUser', user);
+    expect(query).toEqual(`MERGE (user:TGUser {id: $user_id})`);
+    expect(params).toEqual({ user_id: 3 });
+  });
+
+  test('Empty Object', () => {
+    const user = {};
+    const nodeHelper = new NodeHelper();
+    expect(() => nodeHelper.createNode('user', 'TGUser', user)).toThrow(
+      'Object needs one or more keys',
+    );
+  });
+
+  test('Wrong Id', () => {
+    const user = { key: '1234' };
+    const nodeHelper = new NodeHelper();
+    expect(() => nodeHelper.createNode('user', 'TGUser', user)).toThrow(
+      'Object must have the key: id',
+    );
+  });
+});

apps/graph-store/src/cyphers/helpers/node.helper.ts

@@ -0,0 +1,40 @@
+import { Injectable } from '@nestjs/common';
+import { flatten } from 'safe-flat';
+
+@Injectable()
+export class NodeHelper {
+  createNode(
+    name: string,
+    type: 'TGUser' | 'TGChat' | 'TGMessage',
+    obj: any,
+    params: Record<string, unknown> = {},
+    idKey = 'id',
+  ) {
+    if (Object.keys(obj).length < 1) {
+      throw new Error('Object needs one or more keys');
+    } else if (!obj.hasOwnProperty(idKey)) {
+      throw new Error(`Object must have the key: ${idKey}`);
+    }
+
+    const id = obj[idKey];
+    params[`${name}_id`] = id;
+
+    delete obj[idKey];
+    const flat = flatten(obj, '_');
+
+    const setProperties = Object.entries(flat).map(([key, value]) => {
+      params[`${name}_${key}`] = value;
+      return `${name}.${key} = $${name}_${key}`;
+    });
+
+    const query = Object.keys(obj).length
+      ? `
+    MERGE (${name}:${type} {${idKey}: $${name}_id})
+    ON CREATE SET ${setProperties.join(', ')}
+    ON MATCH SET ${setProperties.join(', ')}
+    `
+      : `MERGE (${name}:${type} {${idKey}: $${name}_id})`;
+
+    return { query, params };
+  }
+}

apps/graph-store/src/cyphers/user.cyphers.ts

@@ -0,0 +1,5 @@
+export const CREATE_OR_UPDATE_USER = `
+    MERGE(user:TGUser { id: $userId })
+    ON CREATE SET chat.title = $chatTitle, chat.type = $chatType, chat.timestamp = $timestamp
+    ON MATCH SET chat.title = $chatTitle, chat.type = $chatType, chat.timestamp = $timestamp
+  `;

apps/graph-store/src/decorators/chat.decorator.ts

@@ -0,0 +1,13 @@
+import { createParamDecorator, ExecutionContext } from '@nestjs/common';
+
+export const Chat = createParamDecorator(
+  (data: unknown, ctx: ExecutionContext) => {
+    const { event_data } = ctx.switchToRpc().getData();
+
+    return (
+      event_data.chat ||
+      event_data.edited_message?.chat ||
+      event_data.message_reaction?.chat
+    );
+  },
+);

apps/graph-store/src/decorators/mentions.decorator.ts

@@ -0,0 +1,24 @@
+import { createParamDecorator, ExecutionContext } from '@nestjs/common';
+
+export const Mentions = createParamDecorator(
+  (data: unknown, ctx: ExecutionContext): string[] => {
+    const { event_data } = ctx.switchToRpc().getData();
+
+    const message = event_data || event_data.edited_message;
+
+    const mentions = message.entities?.filter(
+      (entity) => entity.type === 'mention',
+    );
+
+    const usernames = [];
+    if (mentions) {
+      for (const mention of mentions) {
+        usernames.push(
+          message.text.slice(mention.offset, mention.offset + mention.length),
+        );
+      }
+    }
+
+    return usernames;
+  },
+);

apps/graph-store/src/decorators/user.decorator.ts

@@ -0,0 +1,13 @@
+import { createParamDecorator, ExecutionContext } from '@nestjs/common';
+
+export const User = createParamDecorator(
+  (data: unknown, ctx: ExecutionContext) => {
+    const { event_data } = ctx.switchToRpc().getData();
+
+    return (
+      event_data.from ||
+      event_data.edited_message?.from ||
+      event_data.message_reaction?.user
+    );
+  },
+);

apps/graph-store/src/graph-store.controller.ts

@@ -1,12 +1,40 @@
-import { Controller, Get } from '@nestjs/common';
+import { Controller, Inject, Logger, UseInterceptors } from '@nestjs/common';
 import { GraphStoreService } from './graph-store.service';
+import { ClientProxy, MessagePattern } from '@nestjs/microservices';
+import { Events, Services } from '@app/common';
+import { RmqInterceptor } from './interceptors/rmq.interceptor';
+import { User } from './decorators/user.decorator';
+import { Mentions } from './decorators/mentions.decorator';
 
 @Controller()
+@UseInterceptors(RmqInterceptor)
 export class GraphStoreController {
-  constructor(private readonly graphStoreService: GraphStoreService) {}
+  private readonly logger = new Logger(GraphStoreController.name);
 
-  @Get()
-  getHello(): string {
-    return this.graphStoreService.getHello();
+  constructor(
+    private readonly graphStoreService: GraphStoreService,
+    @Inject(Services.Bot) private client: ClientProxy,
+  ) {}
+
+  @MessagePattern(Events.Message)
+  async message(
+    @User() user,
+    @Mentions() mentions,
+    // @Chat() chat,
+  ): Promise<void> {
+    // Get user ids
+    this.client.emit('get_chat', mentions[0]);
+    this.logger.log(user);
+    this.logger.log(mentions);
+  }
+
+  @MessagePattern(Events.EditedMessage)
+  async edited_message(@User() user): Promise<void> {
+    this.logger.log(user);
+  }
+
+  @MessagePattern(Events.MessageReaction)
+  async message_reaction(@User() user): Promise<void> {
+    this.logger.log(user);
   }
 }

apps/graph-store/src/graph-store.module.ts

@@ -1,9 +1,17 @@
 import { Module } from '@nestjs/common';
 import { GraphStoreController } from './graph-store.controller';
 import { GraphStoreService } from './graph-store.service';
-import { RmqModule, neo4jConfig, rmqConfig, schemaConfig } from '@app/common';
+import {
+  Queues,
+  RmqModule,
+  Services,
+  neo4jConfig,
+  rmqConfig,
+  schemaConfig,
+} from '@app/common';
 import { ConfigModule, ConfigService } from '@nestjs/config';
 import { Neo4jModule } from 'nest-neo4j';
+import { NodeHelper } from './cyphers/helpers/node.helper';
 
 @Module({
   imports: [
@@ -13,13 +21,17 @@ import { Neo4jModule } from 'nest-neo4j';
       isGlobal: true,
     }),
     RmqModule,
+    RmqModule.register({
+      name: Services.Bot,
+      queue: Queues.Bot,
+    }),
     Neo4jModule.forRootAsync({
       import: [ConfigModule],
       useFactory: (configService: ConfigService) => configService.get('neo4j'),
       inject: [ConfigService],
     }),
   ],
   controllers: [GraphStoreController],
-  providers: [GraphStoreService],
+  providers: [GraphStoreService, NodeHelper],
 })
 export class GraphStoreModule {}