better note read handling
This commit is contained in:
		
							parent
							
								
									630464f38d
								
							
						
					
					
						commit
						667d58bad4
					
				
					 15 changed files with 109 additions and 66 deletions
				
			
		|  | @ -350,7 +350,8 @@ export default defineComponent({ | |||
| 
 | ||||
| 		capture(withHandler = false) { | ||||
| 			if (this.$i) { | ||||
| 				this.connection.send(document.body.contains(this.$el) ? 'sn' : 's', { id: this.appearNote.id }); | ||||
| 				// TODO: このノートがストリーミング経由で流れてきた場合のみ sr する | ||||
| 				this.connection.send(document.body.contains(this.$el) ? 'sr' : 's', { id: this.appearNote.id }); | ||||
| 				if (withHandler) this.connection.on('noteUpdated', this.onStreamNoteUpdated); | ||||
| 			} | ||||
| 		}, | ||||
|  |  | |||
|  | @ -325,7 +325,8 @@ export default defineComponent({ | |||
| 
 | ||||
| 		capture(withHandler = false) { | ||||
| 			if (this.$i) { | ||||
| 				this.connection.send(document.body.contains(this.$el) ? 'sn' : 's', { id: this.appearNote.id }); | ||||
| 				// TODO: このノートがストリーミング経由で流れてきた場合のみ sr する | ||||
| 				this.connection.send(document.body.contains(this.$el) ? 'sr' : 's', { id: this.appearNote.id }); | ||||
| 				if (withHandler) this.connection.on('noteUpdated', this.onStreamNoteUpdated); | ||||
| 			} | ||||
| 		}, | ||||
|  |  | |||
|  | @ -325,7 +325,8 @@ export default defineComponent({ | |||
| 
 | ||||
| 		capture(withHandler = false) { | ||||
| 			if (this.$i) { | ||||
| 				this.connection.send(document.body.contains(this.$el) ? 'sn' : 's', { id: this.appearNote.id }); | ||||
| 				// TODO: このノートがストリーミング経由で流れてきた場合のみ sr する | ||||
| 				this.connection.send(document.body.contains(this.$el) ? 'sr' : 's', { id: this.appearNote.id }); | ||||
| 				if (withHandler) this.connection.on('noteUpdated', this.onStreamNoteUpdated); | ||||
| 			} | ||||
| 		}, | ||||
|  |  | |||
|  | @ -1,3 +1,5 @@ | |||
| // TODO: 消したい
 | ||||
| 
 | ||||
| const interval = 30 * 60 * 1000; | ||||
| import { AttestationChallenges } from '../models'; | ||||
| import { LessThan } from 'typeorm'; | ||||
|  |  | |||
|  | @ -83,9 +83,7 @@ export default define(meta, async (ps, user) => { | |||
| 
 | ||||
| 	const mentions = await query.take(ps.limit!).getMany(); | ||||
| 
 | ||||
| 	for (const note of mentions) { | ||||
| 		read(user.id, note.id); | ||||
| 	} | ||||
| 	read(user.id, mentions.map(note => note.id)); | ||||
| 
 | ||||
| 	return await Notes.packMany(mentions, user); | ||||
| }); | ||||
|  |  | |||
|  | @ -27,6 +27,8 @@ export default class extends Channel { | |||
| 			// 流れてきたNoteがミュートしているユーザーが関わるものだったら無視する
 | ||||
| 			if (isMutedUserRelated(note, this.muting)) return; | ||||
| 
 | ||||
| 			this.connection.cacheNote(note); | ||||
| 
 | ||||
| 			this.send('note', note); | ||||
| 		} else { | ||||
| 			this.send(type, body); | ||||
|  |  | |||
|  | @ -43,6 +43,8 @@ export default class extends Channel { | |||
| 		// 流れてきたNoteがミュートしているユーザーが関わるものだったら無視する
 | ||||
| 		if (isMutedUserRelated(note, this.muting)) return; | ||||
| 
 | ||||
| 		this.connection.cacheNote(note); | ||||
| 
 | ||||
| 		this.send('note', note); | ||||
| 	} | ||||
| 
 | ||||
|  |  | |||
|  | @ -56,6 +56,8 @@ export default class extends Channel { | |||
| 		// そのためレコードが存在するかのチェックでは不十分なので、改めてcheckWordMuteを呼んでいる
 | ||||
| 		if (this.userProfile && await checkWordMute(note, this.user, this.userProfile.mutedWords)) return; | ||||
| 
 | ||||
| 		this.connection.cacheNote(note); | ||||
| 
 | ||||
| 		this.send('note', note); | ||||
| 	} | ||||
| 
 | ||||
|  |  | |||
|  | @ -37,6 +37,8 @@ export default class extends Channel { | |||
| 		// 流れてきたNoteがミュートしているユーザーが関わるものだったら無視する
 | ||||
| 		if (isMutedUserRelated(note, this.muting)) return; | ||||
| 
 | ||||
| 		this.connection.cacheNote(note); | ||||
| 
 | ||||
| 		this.send('note', note); | ||||
| 	} | ||||
| 
 | ||||
|  |  | |||
|  | @ -64,6 +64,8 @@ export default class extends Channel { | |||
| 		// そのためレコードが存在するかのチェックでは不十分なので、改めてcheckWordMuteを呼んでいる
 | ||||
| 		if (this.userProfile && await checkWordMute(note, this.user, this.userProfile.mutedWords)) return; | ||||
| 
 | ||||
| 		this.connection.cacheNote(note); | ||||
| 
 | ||||
| 		this.send('note', note); | ||||
| 	} | ||||
| 
 | ||||
|  |  | |||
|  | @ -73,6 +73,8 @@ export default class extends Channel { | |||
| 		// そのためレコードが存在するかのチェックでは不十分なので、改めてcheckWordMuteを呼んでいる
 | ||||
| 		if (this.userProfile && await checkWordMute(note, this.user, this.userProfile.mutedWords)) return; | ||||
| 
 | ||||
| 		this.connection.cacheNote(note); | ||||
| 
 | ||||
| 		this.send('note', note); | ||||
| 	} | ||||
| 
 | ||||
|  |  | |||
|  | @ -58,6 +58,8 @@ export default class extends Channel { | |||
| 		// そのためレコードが存在するかのチェックでは不十分なので、改めてcheckWordMuteを呼んでいる
 | ||||
| 		if (this.userProfile && await checkWordMute(note, this.user, this.userProfile.mutedWords)) return; | ||||
| 
 | ||||
| 		this.connection.cacheNote(note); | ||||
| 
 | ||||
| 		this.send('note', note); | ||||
| 	} | ||||
| 
 | ||||
|  |  | |||
|  | @ -18,18 +18,22 @@ export default class extends Channel { | |||
| 				case 'notification': { | ||||
| 					if (this.muting.has(body.userId)) return; | ||||
| 					if (body.note && body.note.isHidden) { | ||||
| 						body.note = await Notes.pack(body.note.id, this.user, { | ||||
| 						const note = await Notes.pack(body.note.id, this.user, { | ||||
| 							detail: true | ||||
| 						}); | ||||
| 						this.connection.cacheNote(note); | ||||
| 						body.note = note; | ||||
| 					} | ||||
| 					break; | ||||
| 				} | ||||
| 				case 'mention': { | ||||
| 					if (this.muting.has(body.userId)) return; | ||||
| 					if (body.isHidden) { | ||||
| 						body = await Notes.pack(body.id, this.user, { | ||||
| 						const note = await Notes.pack(body.id, this.user, { | ||||
| 							detail: true | ||||
| 						}); | ||||
| 						this.connection.cacheNote(note); | ||||
| 						body = note; | ||||
| 					} | ||||
| 					break; | ||||
| 				} | ||||
|  |  | |||
|  | @ -14,6 +14,7 @@ import { AccessToken } from '../../../models/entities/access-token'; | |||
| import { UserProfile } from '../../../models/entities/user-profile'; | ||||
| import { publishChannelStream, publishGroupMessagingStream, publishMessagingStream } from '../../../services/stream'; | ||||
| import { UserGroup } from '../../../models/entities/user-group'; | ||||
| import { PackedNote } from '../../../models/repositories/note'; | ||||
| 
 | ||||
| /** | ||||
|  * Main stream connection | ||||
|  | @ -29,6 +30,7 @@ export default class Connection { | |||
| 	public subscriber: EventEmitter; | ||||
| 	private channels: Channel[] = []; | ||||
| 	private subscribingNotes: any = {}; | ||||
| 	private cachedNotes: PackedNote[] = []; | ||||
| 
 | ||||
| 	constructor( | ||||
| 		wsConnection: websocket.connection, | ||||
|  | @ -115,9 +117,9 @@ export default class Connection { | |||
| 		switch (type) { | ||||
| 			case 'api': this.onApiRequest(body); break; | ||||
| 			case 'readNotification': this.onReadNotification(body); break; | ||||
| 			case 'subNote': this.onSubscribeNote(body, true); break; | ||||
| 			case 'sn': this.onSubscribeNote(body, true); break; // alias
 | ||||
| 			case 's': this.onSubscribeNote(body, false); break; | ||||
| 			case 'subNote': this.onSubscribeNote(body); break; | ||||
| 			case 's': this.onSubscribeNote(body); break; // alias
 | ||||
| 			case 'sr': this.onSubscribeNote(body); this.readNote(body); break; | ||||
| 			case 'unsubNote': this.onUnsubscribeNote(body); break; | ||||
| 			case 'un': this.onUnsubscribeNote(body); break; // alias
 | ||||
| 			case 'connect': this.onChannelConnectRequested(body); break; | ||||
|  | @ -138,6 +140,48 @@ export default class Connection { | |||
| 		this.sendMessageToWs(type, body); | ||||
| 	} | ||||
| 
 | ||||
| 	@autobind | ||||
| 	public cacheNote(note: PackedNote) { | ||||
| 		const add = (note: PackedNote) => { | ||||
| 			const existIndex = this.cachedNotes.findIndex(n => n.id === note.id); | ||||
| 			if (existIndex > -1) { | ||||
| 				this.cachedNotes[existIndex] = note; | ||||
| 				return; | ||||
| 			} | ||||
| 
 | ||||
| 			this.cachedNotes.unshift(note); | ||||
| 			if (this.cachedNotes.length > 32) { | ||||
| 				this.cachedNotes.splice(32); | ||||
| 			} | ||||
| 		}; | ||||
| 
 | ||||
| 		add(note); | ||||
| 		if (note.reply) add(note.reply); | ||||
| 		if (note.renote) add(note.renote); | ||||
| 	} | ||||
| 
 | ||||
| 	@autobind | ||||
| 	private readNote(body: any) { | ||||
| 		const id = body.id; | ||||
| 
 | ||||
| 		const note = this.cachedNotes.find(n => n.id === id); | ||||
| 		if (note == null) return; | ||||
| 
 | ||||
| 		if (this.user && (note.userId !== this.user.id)) { | ||||
| 			if (note.mentions && note.mentions.includes(this.user.id)) { | ||||
| 				readNote(this.user.id, [note]); | ||||
| 			} else if (note.visibleUserIds && note.visibleUserIds.includes(this.user.id)) { | ||||
| 				readNote(this.user.id, [note]); | ||||
| 			} | ||||
| 
 | ||||
| 			if (this.followingChannels.has(note.channelId)) { | ||||
| 				// TODO
 | ||||
| 			} | ||||
| 
 | ||||
| 			// TODO: アンテナの既読処理
 | ||||
| 		} | ||||
| 	} | ||||
| 
 | ||||
| 	/** | ||||
| 	 * APIリクエスト要求時 | ||||
| 	 */ | ||||
|  | @ -174,7 +218,7 @@ export default class Connection { | |||
| 	 * 投稿購読要求時 | ||||
| 	 */ | ||||
| 	@autobind | ||||
| 	private onSubscribeNote(payload: any, read: boolean) { | ||||
| 	private onSubscribeNote(payload: any) { | ||||
| 		if (!payload.id) return; | ||||
| 
 | ||||
| 		if (this.subscribingNotes[payload.id] == null) { | ||||
|  | @ -186,12 +230,6 @@ export default class Connection { | |||
| 		if (this.subscribingNotes[payload.id] === 1) { | ||||
| 			this.subscriber.on(`noteStream:${payload.id}`, this.onNoteStreamMessage); | ||||
| 		} | ||||
| 
 | ||||
| 		if (this.user && read) { | ||||
| 			// TODO: クライアントでタイムライン読み込みなどすると、一度に大量のreadNoteが発生しクエリ数がすごいことになるので、ある程度まとめてreadNoteするようにする
 | ||||
| 			// 具体的には、この箇所ではキュー的な配列にread予定ノートを溜めておくに留めて、別の箇所で定期的にキューにあるノートを配列でreadNoteに渡すような実装にする
 | ||||
| 			readNote(this.user.id, payload.id); | ||||
| 		} | ||||
| 	} | ||||
| 
 | ||||
| 	/** | ||||
|  |  | |||
|  | @ -2,70 +2,54 @@ import { publishMainStream } from '../stream'; | |||
| import { Note } from '../../models/entities/note'; | ||||
| import { User } from '../../models/entities/user'; | ||||
| import { NoteUnreads, Antennas, AntennaNotes, Users } from '../../models'; | ||||
| import { Not, IsNull } from 'typeorm'; | ||||
| import { Not, IsNull, In } from 'typeorm'; | ||||
| 
 | ||||
| /** | ||||
|  * Mark a note as read | ||||
|  * Mark notes as read | ||||
|  */ | ||||
| export default async function( | ||||
| 	userId: User['id'], | ||||
| 	noteId: Note['id'] | ||||
| 	noteIds: Note['id'][] | ||||
| ) { | ||||
| 	async function careNoteUnreads() { | ||||
| 		const exist = await NoteUnreads.findOne({ | ||||
| 			userId: userId, | ||||
| 			noteId: noteId, | ||||
| 		}); | ||||
| 
 | ||||
| 		if (!exist) return; | ||||
| 
 | ||||
| 		// Remove the record
 | ||||
| 		await NoteUnreads.delete({ | ||||
| 			userId: userId, | ||||
| 			noteId: noteId, | ||||
| 			noteId: In(noteIds), | ||||
| 		}); | ||||
| 
 | ||||
| 		if (exist.isMentioned) { | ||||
| 			NoteUnreads.count({ | ||||
| 				userId: userId, | ||||
| 				isMentioned: true | ||||
| 			}).then(mentionsCount => { | ||||
| 				if (mentionsCount === 0) { | ||||
| 					// 全て既読になったイベントを発行
 | ||||
| 					publishMainStream(userId, 'readAllUnreadMentions'); | ||||
| 				} | ||||
| 			}); | ||||
| 		} | ||||
| 		NoteUnreads.count({ | ||||
| 			userId: userId, | ||||
| 			isMentioned: true | ||||
| 		}).then(mentionsCount => { | ||||
| 			if (mentionsCount === 0) { | ||||
| 				// 全て既読になったイベントを発行
 | ||||
| 				publishMainStream(userId, 'readAllUnreadMentions'); | ||||
| 			} | ||||
| 		}); | ||||
| 
 | ||||
| 		if (exist.isSpecified) { | ||||
| 			NoteUnreads.count({ | ||||
| 				userId: userId, | ||||
| 				isSpecified: true | ||||
| 			}).then(specifiedCount => { | ||||
| 				if (specifiedCount === 0) { | ||||
| 					// 全て既読になったイベントを発行
 | ||||
| 					publishMainStream(userId, 'readAllUnreadSpecifiedNotes'); | ||||
| 				} | ||||
| 			}); | ||||
| 		} | ||||
| 		NoteUnreads.count({ | ||||
| 			userId: userId, | ||||
| 			isSpecified: true | ||||
| 		}).then(specifiedCount => { | ||||
| 			if (specifiedCount === 0) { | ||||
| 				// 全て既読になったイベントを発行
 | ||||
| 				publishMainStream(userId, 'readAllUnreadSpecifiedNotes'); | ||||
| 			} | ||||
| 		}); | ||||
| 
 | ||||
| 		if (exist.noteChannelId) { | ||||
| 			NoteUnreads.count({ | ||||
| 				userId: userId, | ||||
| 				noteChannelId: Not(IsNull()) | ||||
| 			}).then(channelNoteCount => { | ||||
| 				if (channelNoteCount === 0) { | ||||
| 					// 全て既読になったイベントを発行
 | ||||
| 					publishMainStream(userId, 'readAllChannels'); | ||||
| 				} | ||||
| 			}); | ||||
| 		} | ||||
| 		NoteUnreads.count({ | ||||
| 			userId: userId, | ||||
| 			noteChannelId: Not(IsNull()) | ||||
| 		}).then(channelNoteCount => { | ||||
| 			if (channelNoteCount === 0) { | ||||
| 				// 全て既読になったイベントを発行
 | ||||
| 				publishMainStream(userId, 'readAllChannels'); | ||||
| 			} | ||||
| 		}); | ||||
| 	} | ||||
| 
 | ||||
| 	async function careAntenna() { | ||||
| 		const beforeUnread = await Users.getHasUnreadAntenna(userId); | ||||
| 		if (!beforeUnread) return; | ||||
| 
 | ||||
| 		const antennas = await Antennas.find({ userId }); | ||||
| 
 | ||||
| 		await Promise.all(antennas.map(async antenna => { | ||||
|  | @ -78,7 +62,7 @@ export default async function( | |||
| 
 | ||||
| 			await AntennaNotes.update({ | ||||
| 				antennaId: antenna.id, | ||||
| 				noteId: noteId | ||||
| 				noteId: In(noteIds) | ||||
| 			}, { | ||||
| 				read: true | ||||
| 			}); | ||||
|  |  | |||
		Loading…
	
	Add table
		Add a link
		
	
		Reference in a new issue