Merge branch 'dev'

.gitattributes

@@ -1,2 +1,3 @@
 src/config/** merge=ours
-ecosystem.config.js merge=ours
+ecosystem.config.js merge=ours
+update.sh merge=ours

package.json

@@ -16,6 +16,7 @@
     "@midwayjs/swagger": "^3.9.9",
     "@midwayjs/typegoose": "^3.9.0",
     "@midwayjs/validate": "^3.0.0",
+    "@midwayjs/ws": "^3.9.0",
     "@typegoose/typegoose": "^10.0.0",
     "amqp-connection-manager": "^4.1.10",
     "amqplib": "^0.10.3",
@@ -33,6 +34,7 @@
     "@types/koa": "^2.13.4",
     "@types/lodash": "^4.14.191",
     "@types/node": "14",
+    "@types/ws": "^8.5.4",
     "cross-env": "^6.0.0",
     "jest": "^29.2.2",
     "mwts": "^1.0.5",

src/config/config.local.ts

@@ -8,6 +8,9 @@ export default {
     port: 12214,
     globalPrefix: '/dev/point/chat/v1/api',
   },
+  webSocket: {
+    clientTracking: true,
+  },
   swagger: {
     swaggerPath: '/dev/point/chat/v1/api/doc/api',
   },
@@ -23,6 +26,16 @@ export default {
         },
         entities: ['./entity/chat'],
       },
+      base: {
+        uri: `mongodb://${ip}:27017/point_shopping-dev`,
+        options: {
+          user: 'admin',
+          pass: 'admin',
+          authSource: 'admin',
+          useNewUrlParser: true,
+        },
+        entities: ['./entity/base'],
+      },
     },
   },
   redis: {

src/configuration.ts

@@ -11,6 +11,7 @@ import * as axios from '@midwayjs/axios';
 import { Inject } from '@midwayjs/core';
 import * as rabbitmq from '@midwayjs/rabbitmq';
 import { Context } from '@midwayjs/koa';
+import * as ws from '@midwayjs/ws';
 @Configuration({
   imports: [
     FreeFrame,
@@ -19,6 +20,7 @@ import { Context } from '@midwayjs/koa';
     axios,
     swagger,
     rabbitmq,
+    ws,
     {
       component: info,
       enabledEnvironment: ['local'],

src/controller/chatRecord.controller.ts

@@ -19,6 +19,9 @@ import * as dayjs from 'dayjs';
 import { RoomService } from '../service/room.service';
 import get = require('lodash/get');
 import { MqSender } from '../service/mq/mqSender.service';
+import { WsSocketController } from '../socket/ws.controller';
+import { Types } from 'mongoose';
+const ObjectId = Types.ObjectId;
 @ApiTags(['聊天记录'])
 @Controller('/chatRecord')
 export class ChatRecordController extends BaseController {
@@ -30,6 +33,10 @@ export class ChatRecordController extends BaseController {
 
   @Inject()
   mqSender: MqSender;
+
+  @Inject()
+  ws: WsSocketController;
+
   @Post('/')
   @Validate()
   @ApiResponse({ type: CreateVO_chatRecord })
@@ -55,28 +62,26 @@ export class ChatRecordController extends BaseController {
       if (roomId) {
         data.room = roomId;
       } else {
-        roomData.last_chat = data.content;
-        roomData.last_person = speaker;
-        roomData.last_time = data.time;
         const roomDbData = await this.roomService.create(roomData);
         if (!roomDbData) throw new ServiceError('房间创建失败', FrameworkErrorEnum.SERVICE_FAULT);
         data.room = get(roomDbData, '_id');
       }
       result = await this.service.create(data);
     }
-    // mq 发送消息
+    const rd = { last_chat: get(result, '_id') };
+    await this.roomService.updateOne(data.room, rd);
     const room = await this.roomService.fetch(data.room);
     let receiver;
-    if (get(room, 'customer') === get(data, 'speaker')) receiver = get(room, 'shop');
-    else receiver = get(room, 'customer');
-    await this.mqSender.toSendMsg(receiver, data);
+    if (new ObjectId(get(room, 'customer._id')).equals(get(data, 'speaker'))) receiver = get(room, 'shop._id');
+    else receiver = get(room, 'customer._id');
+    await this.ws.toSend({ ...JSON.parse(JSON.stringify(result)), type: 'chat' }, receiver);
     return result;
   }
   @Get('/')
   @ApiQuery({ name: 'query' })
   @ApiResponse({ type: QueryVO_chatRecord })
   async query(@Query() filter: QueryDTO_chatRecord, @Query('skip') skip: number, @Query('limit') limit: number) {
-    const data = await this.service.query(filter, { skip, limit });
+    const data = await this.service.query(filter, { skip, limit, sort: { 'meta.createdAt': -1 } });
     const total = await this.service.count(filter);
     return { data, total };
   }

src/controller/home.controller.ts

@@ -1,9 +1,19 @@
-import { Controller, Get } from '@midwayjs/decorator';
-
+import { Body, Controller, Get, Inject, Post } from '@midwayjs/decorator';
+import { SendMsgDTO } from '../interface/websocket.interface';
+import { WsSocketController } from '../socket/ws.controller';
 @Controller('/')
 export class HomeController {
+  @Inject()
+  ws: WsSocketController;
   @Get('/')
   async home(): Promise<string> {
     return 'Hello Midwayjs!';
   }
+
+  @Post('/sendWebSocket')
+  async sendWebSocket(@Body() object: SendMsgDTO) {
+    const { recevier, ...others } = object;
+    await this.ws.toSend(others, recevier);
+    return 'ok';
+  }
 }

src/controller/room.controller.ts

@@ -21,9 +21,14 @@ export class RoomController extends BaseController {
   @ApiQuery({ name: 'query' })
   @ApiResponse({ type: QueryVO_room })
   async query(@Query() filter: QueryDTO_room, @Query('skip') skip: number, @Query('limit') limit: number) {
-    const data = await this.service.query(filter, { skip, limit });
-    const total = await this.service.count(filter);
-    return { data, total };
+    const { data, total } = await this.service.newQuery(filter, { skip, limit, sort: { 'meta.createdAt': -1 } });
+    const list = [];
+    for (const i of data) {
+      let nd = await this.service.fillId(i);
+      nd = await this.service.countNotRead(filter, nd);
+      list.push(nd);
+    }
+    return { data: list, total };
   }
 
   @Get('/:id')

src/entity/base/shop.ts

@@ -0,0 +1,43 @@
+import { modelOptions, prop } from '@typegoose/typegoose';
+import { BaseModel } from 'free-midway-component';
+@modelOptions({
+  schemaOptions: { collection: 'shop' },
+})
+export class Shop extends BaseModel {
+  @prop({ required: false, index: false, zh: '店铺logo' })
+  logo: Array<any>;
+  @prop({ required: false, index: true, zh: '商店名称' })
+  name: string;
+  @prop({
+    required: false,
+    index: true,
+    zh: '店铺编号',
+    remark: '自增,中间件处理',
+  })
+  code: string;
+  @prop({ required: false, index: true, zh: '店主' })
+  person: string;
+  @prop({ required: false, index: true, zh: '联系电话' })
+  phone: string;
+  @prop({ required: false, index: false, zh: '地址' })
+  address: string;
+  @prop({ required: false, index: false, zh: '证件照片' })
+  file: Array<any>;
+  @prop({
+    required: false,
+    index: true,
+    zh: '店铺状态',
+    remark: '字典:shop_status',
+  })
+  status: string;
+  @prop({ required: false, index: false, zh: '商品评分', default: '0' })
+  goods_score: number;
+  @prop({ required: false, index: false, zh: '发货评分', default: '0' })
+  send_score: number;
+  @prop({ required: false, index: false, zh: '服务评分', default: '0' })
+  service_score: number;
+  @prop({ required: false, index: false, zh: '抽成比例', default: 0 })
+  cut: number;
+  @prop({ required: false, index: false, zh: '二维码' })
+  qrcode: Array<any>;
+}

src/entity/base/user.ts

@@ -0,0 +1,39 @@
+import { modelOptions, prop } from '@typegoose/typegoose';
+import { BaseModel } from 'free-midway-component';
+@modelOptions({
+  schemaOptions: { collection: 'user' },
+})
+export class User extends BaseModel {
+  @prop({ required: false, index: true, zh: '用户名' })
+  name: string;
+  @prop({ required: false, index: true, zh: '手机号' })
+  phone: string;
+  @prop({ required: false, index: true, zh: '电子邮箱' })
+  email: string;
+  @prop({ required: false, index: false, zh: '头像' })
+  icon: Array<any>;
+  @prop({ required: false, index: false, zh: '生日' })
+  birth: string;
+  @prop({ required: false, index: true, zh: '性别', remark: '字典:gender' })
+  gender: string;
+  @prop({ required: false, index: true, zh: '微信小程序' })
+  openid: string;
+  @prop({
+    required: false,
+    index: true,
+    zh: '状态',
+    remark: '字典:user_status',
+    default: '0',
+  })
+  status: string;
+  @prop({
+    required: false,
+    index: true,
+    zh: '是否是团长',
+    remark: '字典:is_use',
+    default: '1',
+  })
+  is_leader: string;
+  @prop({ required: false, index: true, zh: '密码', select: false })
+  password: object;
+}

src/entity/chat/needSend.entity.ts

@@ -0,0 +1,11 @@
+import { modelOptions, prop } from '@typegoose/typegoose';
+import { BaseModel } from 'free-midway-component';
+@modelOptions({
+  schemaOptions: { collection: 'needSend' },
+})
+export class NeedSend extends BaseModel {
+  @prop({ required: false, index: true, zh: '接收人' })
+  to: string;
+  @prop({ required: false, index: false, zh: '消息' })
+  msg: object;
+}

src/entity/chat/room.entity.ts

@@ -4,14 +4,10 @@ import { BaseModel } from 'free-midway-component';
   schemaOptions: { collection: 'room' },
 })
 export class Room extends BaseModel {
-  @prop({ required: false, index: true, zh: '顾客' })
+  @prop({ required: false, index: true, zh: '顾客', ref: 'User' })
   customer: string;
-  @prop({ required: false, index: true, zh: '店铺' })
+  @prop({ required: false, index: true, zh: '店铺', ref: 'Shop' })
   shop: string;
-  @prop({ required: false, index: false, zh: '最后发言' })
+  @prop({ required: false, index: false, zh: '最后发言', ref: 'ChatRecord' })
   last_chat: string;
-  @prop({ required: false, index: false, zh: '最后发言人' })
-  last_person: string;
-  @prop({ required: false, index: false, zh: '最后发言时间' })
-  last_time: string;
 }

src/interface/chatRecord.interface.ts

@@ -27,7 +27,7 @@ export class FetchVO_chatRecord {
 export class QueryDTO_chatRecord extends SearchBase {
   constructor() {
     const like_prop = [];
-    const props = ['room', 'speaker', 'time'];
+    const props = ['room', 'speaker', 'time', 'is_read'];
     const mapping = [];
     super({ like_prop, props, mapping });
   }
@@ -37,6 +37,8 @@ export class QueryDTO_chatRecord extends SearchBase {
   'speaker': string = undefined;
   @ApiProperty({ description: '时间' })
   'time': string = undefined;
+  @ApiProperty({ description: '是否已读' })
+  'is_read': string = undefined;
 }
 
 export class QueryVO_chatRecord extends FetchVO_chatRecord {}

src/interface/room.interface.ts

@@ -16,10 +16,6 @@ export class FetchVO_room {
   'shop': string = undefined;
   @ApiProperty({ description: '最后发言' })
   'last_chat': string = undefined;
-  @ApiProperty({ description: '最后发言人' })
-  'last_person': string = undefined;
-  @ApiProperty({ description: '最后发言时间' })
-  'last_time': string = undefined;
 }
 
 export class QueryDTO_room extends SearchBase {
@@ -47,12 +43,6 @@ export class CreateDTO_room {
   @ApiProperty({ description: '最后发言' })
   @Rule(RuleType['string']().empty(''))
   'last_chat': string = undefined;
-  @ApiProperty({ description: '最后发言人' })
-  @Rule(RuleType['string']().empty(''))
-  'last_person': string = undefined;
-  @ApiProperty({ description: '最后发言时间' })
-  @Rule(RuleType['string']().empty(''))
-  'last_time': string = undefined;
 }
 
 export class CreateVO_room extends FetchVO_room {}

src/interface/websocket.interface.ts

@@ -0,0 +1,11 @@
+import { ApiProperty } from '@midwayjs/swagger';
+import { Rule, RuleType } from '@midwayjs/validate';
+
+export class SendMsgDTO {
+  @ApiProperty({ description: '接收人' })
+  @Rule(RuleType['string']().required())
+  'recevier': string = undefined;
+  @ApiProperty({ description: '消息类型' })
+  @Rule(RuleType['string']().required())
+  'type': string = undefined;
+}

src/service/mq/mqSender.service.ts

@@ -29,13 +29,13 @@ export class MqSender {
    * @param data 数据
    */
   async toSendMsg(receiver, data) {
-    await this.connect(receiver);
-    const res = await this.sendMsg(receiver, data);
-    console.log(res);
+    await this.connect(receiver, data);
+    // const res = await this.sendMsg(receiver, data);
+    // console.log(res);
     await this.close();
   }
 
-  async connect(queue) {
+  async connect(queue, data) {
     // 创建连接，你可以把配置放在 Config 中，然后注入进来
     this.connection = await amqp.connect(this.mqUrl);
     // 创建 channel
@@ -47,6 +47,7 @@ export class MqSender {
           channel.assertExchange(_.get(this.mqConfig, 'ex'), 'direct', { durable: true }),
           channel.assertQueue(queue, { durable: false, exclusive: false }),
           channel.bindQueue(queue, _.get(this.mqConfig, 'ex')),
+          channel.publish(_.get(this.mqConfig, 'ex'), queue, { ...data, type: 'chat' })
         ]);
       },
     });

src/service/room.service.ts

@@ -1,17 +1,83 @@
 import { Provide } from '@midwayjs/decorator';
 import { InjectEntityModel } from '@midwayjs/typegoose';
 import { ReturnModelType } from '@typegoose/typegoose';
-import { BaseService } from 'free-midway-component';
+import { BaseService, PageOptions, SearchBase } from 'free-midway-component';
 import { Room } from '../entity/chat/room.entity';
 import get = require('lodash/get');
+import cloneDeep = require('lodash/cloneDeep');
+import head = require('lodash/head');
+import { ChatRecord } from '../entity/chat/chatRecord.entity';
+import { User } from '../entity/base/user';
+import { Shop } from '../entity/base/shop';
+
 type modelType = ReturnModelType<typeof Room>;
 @Provide()
 export class RoomService extends BaseService<modelType> {
   @InjectEntityModel(Room)
   model: modelType;
+  @InjectEntityModel(ChatRecord)
+  chatRecordModel: ReturnModelType<typeof ChatRecord>;
+
+  @InjectEntityModel(User)
+  userModel: ReturnModelType<typeof User>;
+
+  @InjectEntityModel(Shop)
+  shopModel: ReturnModelType<typeof Shop>;
 
   async checkRoomIsExist(query) {
     const data = await this.model.findOne(query, { _id: 1 });
     return get(data, '_id');
   }
+
+  // 查询未读
+  async countNotRead(filter, data) {
+    const query: any = { room: data._id, is_read: '0' };
+    if (filter.customer) query.speaker = get(data, 'shop._id');
+    else if (filter.shop) query.speaker = get(data, 'customer._id');
+    else return data;
+    const num = await this.chatRecordModel.count(query);
+    data.not_read = num;
+    return data;
+  }
+  // 填充 顾客和商品id对应的数据
+  async fillId(data) {
+    const user_id = get(data, 'customer');
+    const shop_id = get(data, 'shop');
+    const user = await this.userModel.findById(user_id, { meta: 0, __v: 0 }).lean();
+    const shop = await this.shopModel.findById(shop_id, { meta: 0, __v: 0 }).lean();
+    data.customer = user;
+    data.shop = shop;
+    return data;
+  }
+
+  async newQuery(filter: SearchBase, pageOptions: PageOptions) {
+    const pipeline = [];
+    const dup = cloneDeep(filter.getFilter());
+    pipeline.push({ $match: dup });
+    pipeline.push({
+      $addFields: {
+        last_chat_oid: {
+          $toObjectId: '$last_chat',
+        },
+        last_chat_id: '$last_chat',
+      },
+    });
+    pipeline.push({
+      $lookup: {
+        from: 'chatRecord',
+        localField: 'last_chat_oid',
+        foreignField: '_id',
+        as: 'last_chat',
+      },
+    });
+    pipeline.push({ $unwind: '$last_chat' });
+    pipeline.push({ $sort: { 'last_chat.meta.createdAt': -1 } });
+    const qp = cloneDeep(pipeline);
+    if (get(pageOptions, 'skip')) qp.push({ $skip: get(pageOptions, 'skip') });
+    if (get(pageOptions, 'limit')) qp.push({ $limit: get(pageOptions, 'limit') });
+    const data = await this.model.aggregate(qp);
+    const totalResult = await this.model.aggregate([...pipeline, { $count: 'total' }]);
+    const total = get(head(totalResult), 'total', 0);
+    return { data, total };
+  }
 }

src/socket/ws.controller.ts

@@ -0,0 +1,98 @@
+import { App, Inject, OnWSConnection, OnWSDisConnection, OnWSMessage, WSController } from '@midwayjs/decorator';
+import { Context } from '@midwayjs/ws';
+import * as http from 'http';
+import get = require('lodash/get');
+import groupBy = require('lodash/groupBy');
+import { Application } from '@midwayjs/ws';
+import last = require('lodash/last');
+import { Types } from 'mongoose';
+const ObjectId = Types.ObjectId;
+import { NeedSend } from '../entity/chat/needSend.entity';
+import { ReturnModelType } from '@typegoose/typegoose';
+import { InjectEntityModel } from '@midwayjs/typegoose';
+
+@WSController('*/ws/*')
+export class WsSocketController {
+  @Inject()
+  ctx: Context;
+  @App('webSocket')
+  wsApp: Application;
+  @InjectEntityModel(NeedSend)
+  model: ReturnModelType<typeof NeedSend>;
+
+  @OnWSConnection()
+  async onConnectionMethod(socket: Context, request: http.IncomingMessage) {
+    const url = get(request, 'url');
+    const arr = url.split('/');
+    const token = last(arr);
+    socket.send('connect success');
+    const acs = this.getClients();
+    // 最后一个是该websocket实例,赋上token
+    const client = last(acs);
+    client.token = token;
+    await this.checkNeedSend(client);
+  }
+  /**
+   * 给对象补发 未发送出去的消息
+   * @param client ws连接实例
+   */
+  async checkNeedSend(client) {
+    const to = get(client, 'token');
+    if (!to) return;
+    let list = await this.model.find({ to }).lean();
+    list = list.map(i => ({ ...i, type: get(i, 'msg.type') }));
+    const groups = groupBy(list, 'type');
+    for (const key in groups) {
+      const list = groups[key];
+      const type = get(last(list), 'type');
+      await this.toSend({ type }, to);
+    }
+    await this.model.deleteMany({ to });
+  }
+
+  // 获取信息
+  @OnWSMessage('message')
+  async gotMessage(data: Buffer) {
+    // const msg = data.toString();
+    return '(╯‵□′)╯︵┻━┻';
+  }
+
+  @OnWSDisConnection()
+  async disconnect(id) {
+    // 断开连接
+  }
+  /**
+   * 给指定对象发送消息
+   * @param data 要发送的数据
+   * @param token 店铺id/用户id
+   */
+  async toSend(data: object, token) {
+    const clients = this.getClient(token);
+    if (!clients) {
+      // 存储到待发送表中
+      await this.model.create({ to: token, msg: data });
+      return;
+    }
+    clients.send(JSON.stringify(data));
+  }
+  /**
+   * 获取所有连接实例
+   */
+  getClients(): Array<any> {
+    const acs = [];
+    const clients = this.wsApp.clients;
+    clients.forEach(e => {
+      acs.push(e);
+    });
+    return acs;
+  }
+  /**
+   * 获取指定实例连接
+   * @param token 用户id/店铺id
+   */
+  getClient(token: string) {
+    const clients = this.getClients();
+    const client = clients.find(f => new ObjectId(f.token).equals(token));
+    return client;
+  }
+}

update.sh

@@ -0,0 +1 @@
+git pull && npm run build && pm2 restart 36