1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
import { Inject, Injectable } from '@nestjs/common';
import type { UserGroupJoiningsRepository, UsersRepository, MessagingMessagesRepository } from '@/models/index.js';
import type { User, ILocalUser, IRemoteUser } from '@/models/entities/User.js';
import type { UserGroup } from '@/models/entities/UserGroup.js';
import { MessagingService } from '@/core/MessagingService.js';
import { UserEntityService } from '@/core/entities/UserEntityService.js';
import { DI } from '@/di-symbols.js';
import Channel from '../channel.js';
import type { StreamMessages } from '../types.js';
class MessagingChannel extends Channel {
public readonly chName = 'messaging';
public static shouldShare = false;
public static requireCredential = true;
private otherpartyId: string | null;
private otherparty: User | null;
private groupId: string | null;
private subCh: `messagingStream:${User['id']}-${User['id']}` | `messagingStream:${UserGroup['id']}`;
private typers: Record<User['id'], Date> = {};
private emitTypersIntervalId: ReturnType<typeof setInterval>;
constructor(
private usersRepository: UsersRepository,
private userGroupJoiningsRepository: UserGroupJoiningsRepository,
private messagingMessagesRepository: MessagingMessagesRepository,
private userEntityService: UserEntityService,
private messagingService: MessagingService,
id: string,
connection: Channel['connection'],
) {
super(id, connection);
this.onEvent = this.onEvent.bind(this);
this.onMessage = this.onMessage.bind(this);
this.emitTypers = this.emitTypers.bind(this);
}
public async init(params: any) {
this.otherpartyId = params.otherparty;
this.otherparty = this.otherpartyId ? await this.usersRepository.findOneByOrFail({ id: this.otherpartyId }) : null;
this.groupId = params.group;
// Check joining
if (this.groupId) {
const joining = await this.userGroupJoiningsRepository.findOneBy({
userId: this.user!.id,
userGroupId: this.groupId,
});
if (joining == null) {
return;
}
}
this.emitTypersIntervalId = setInterval(this.emitTypers, 5000);
this.subCh = this.otherpartyId
? `messagingStream:${this.user!.id}-${this.otherpartyId}`
: `messagingStream:${this.groupId}`;
// Subscribe messaging stream
this.subscriber.on(this.subCh, this.onEvent);
}
private onEvent(data: StreamMessages['messaging']['payload'] | StreamMessages['groupMessaging']['payload']) {
if (data.type === 'typing') {
const id = data.body;
const begin = this.typers[id] == null;
this.typers[id] = new Date();
if (begin) {
this.emitTypers();
}
} else {
this.send(data);
}
}
public onMessage(type: string, body: any) {
switch (type) {
case 'read':
if (this.otherpartyId) {
this.messagingService.readUserMessagingMessage(this.user!.id, this.otherpartyId, [body.id]);
// リモートユーザーからのメッセージだったら既読配信
if (this.userEntityService.isLocalUser(this.user!) && this.userEntityService.isRemoteUser(this.otherparty!)) {
this.messagingMessagesRepository.findOneBy({ id: body.id }).then(message => {
if (message) this.messagingService.deliverReadActivity(this.user as ILocalUser, this.otherparty as IRemoteUser, message);
});
}
} else if (this.groupId) {
this.messagingService.readGroupMessagingMessage(this.user!.id, this.groupId, [body.id]);
}
break;
}
}
private async emitTypers() {
const now = new Date();
// Remove not typing users
for (const [userId, date] of Object.entries(this.typers)) {
if (now.getTime() - date.getTime() > 5000) delete this.typers[userId];
}
const users = await this.userEntityService.packMany(Object.keys(this.typers), null, { detail: false });
this.send({
type: 'typers',
body: users,
});
}
public dispose() {
this.subscriber.off(this.subCh, this.onEvent);
clearInterval(this.emitTypersIntervalId);
}
}
@Injectable()
export class MessagingChannelService {
public readonly shouldShare = MessagingChannel.shouldShare;
public readonly requireCredential = MessagingChannel.requireCredential;
constructor(
@Inject(DI.usersRepository)
private usersRepository: UsersRepository,
@Inject(DI.userGroupJoiningsRepository)
private userGroupJoiningsRepository: UserGroupJoiningsRepository,
@Inject(DI.messagingMessagesRepository)
private messagingMessagesRepository: MessagingMessagesRepository,
private userEntityService: UserEntityService,
private messagingService: MessagingService,
) {
}
public create(id: string, connection: Channel['connection']): MessagingChannel {
return new MessagingChannel(
this.usersRepository,
this.userGroupJoiningsRepository,
this.messagingMessagesRepository,
this.userEntityService,
this.messagingService,
id,
connection,
);
}
}
|
