diff --git a/src/client/app/common/mios.ts b/src/client/app/common/mios.ts index 96a04bad3..6d6d6b3e6 100644 --- a/src/client/app/common/mios.ts +++ b/src/client/app/common/mios.ts @@ -9,11 +9,12 @@ import Connection from './scripts/streaming/stream'; import { HomeStreamManager } from './scripts/streaming/home'; import { DriveStreamManager } from './scripts/streaming/drive'; import { ServerStreamManager } from './scripts/streaming/server'; -import { RequestsStreamManager } from './scripts/streaming/requests'; import { MessagingIndexStreamManager } from './scripts/streaming/messaging-index'; import { OthelloStreamManager } from './scripts/streaming/othello'; import Err from '../common/views/components/connect-failed.vue'; +import { LocalTimelineStreamManager } from './scripts/streaming/local-timeline'; +import { GlobalTimelineStreamManager } from './scripts/streaming/global-timeline'; //#region api requests let spinner = null; @@ -116,15 +117,17 @@ export default class MiOS extends EventEmitter { * Connection managers */ public streams: { + localTimelineStream: LocalTimelineStreamManager; + globalTimelineStream: GlobalTimelineStreamManager; driveStream: DriveStreamManager; serverStream: ServerStreamManager; - requestsStream: RequestsStreamManager; messagingIndexStream: MessagingIndexStreamManager; othelloStream: OthelloStreamManager; } = { + localTimelineStream: null, + globalTimelineStream: null, driveStream: null, serverStream: null, - requestsStream: null, messagingIndexStream: null, othelloStream: null }; @@ -231,13 +234,14 @@ export default class MiOS extends EventEmitter { public async init(callback) { //#region Init stream managers this.streams.serverStream = new ServerStreamManager(this); - this.streams.requestsStream = new RequestsStreamManager(this); this.once('signedin', () => { // Init home stream manager this.stream = new HomeStreamManager(this, this.i); // Init other stream manager + this.streams.localTimelineStream = new LocalTimelineStreamManager(this, this.i); + this.streams.globalTimelineStream = new GlobalTimelineStreamManager(this, this.i); this.streams.driveStream = new DriveStreamManager(this, this.i); this.streams.messagingIndexStream = new MessagingIndexStreamManager(this, this.i); this.streams.othelloStream = new OthelloStreamManager(this, this.i); diff --git a/src/client/app/common/scripts/streaming/global-timeline.ts b/src/client/app/common/scripts/streaming/global-timeline.ts new file mode 100644 index 000000000..452ddbac0 --- /dev/null +++ b/src/client/app/common/scripts/streaming/global-timeline.ts @@ -0,0 +1,34 @@ +import Stream from './stream'; +import StreamManager from './stream-manager'; +import MiOS from '../../mios'; + +/** + * Global timeline stream connection + */ +export class GlobalTimelineStream extends Stream { + constructor(os: MiOS, me) { + super(os, 'global-timeline', { + i: me.token + }); + } +} + +export class GlobalTimelineStreamManager extends StreamManager { + private me; + private os: MiOS; + + constructor(os: MiOS, me) { + super(); + + this.me = me; + this.os = os; + } + + public getConnection() { + if (this.connection == null) { + this.connection = new GlobalTimelineStream(this.os, this.me); + } + + return this.connection; + } +} diff --git a/src/client/app/common/scripts/streaming/local-timeline.ts b/src/client/app/common/scripts/streaming/local-timeline.ts new file mode 100644 index 000000000..3d04e05cd --- /dev/null +++ b/src/client/app/common/scripts/streaming/local-timeline.ts @@ -0,0 +1,34 @@ +import Stream from './stream'; +import StreamManager from './stream-manager'; +import MiOS from '../../mios'; + +/** + * Local timeline stream connection + */ +export class LocalTimelineStream extends Stream { + constructor(os: MiOS, me) { + super(os, 'local-timeline', { + i: me.token + }); + } +} + +export class LocalTimelineStreamManager extends StreamManager { + private me; + private os: MiOS; + + constructor(os: MiOS, me) { + super(); + + this.me = me; + this.os = os; + } + + public getConnection() { + if (this.connection == null) { + this.connection = new LocalTimelineStream(this.os, this.me); + } + + return this.connection; + } +} diff --git a/src/client/app/common/scripts/streaming/requests.ts b/src/client/app/common/scripts/streaming/requests.ts deleted file mode 100644 index 5bec30143..000000000 --- a/src/client/app/common/scripts/streaming/requests.ts +++ /dev/null @@ -1,30 +0,0 @@ -import Stream from './stream'; -import StreamManager from './stream-manager'; -import MiOS from '../../mios'; - -/** - * Requests stream connection - */ -export class RequestsStream extends Stream { - constructor(os: MiOS) { - super(os, 'requests'); - } -} - -export class RequestsStreamManager extends StreamManager { - private os: MiOS; - - constructor(os: MiOS) { - super(); - - this.os = os; - } - - public getConnection() { - if (this.connection == null) { - this.connection = new RequestsStream(this.os); - } - - return this.connection; - } -} diff --git a/src/client/app/desktop/views/components/timeline.core.vue b/src/client/app/desktop/views/components/timeline.core.vue new file mode 100644 index 000000000..8056e5b2c --- /dev/null +++ b/src/client/app/desktop/views/components/timeline.core.vue @@ -0,0 +1,175 @@ + + + + + diff --git a/src/client/app/desktop/views/components/timeline.vue b/src/client/app/desktop/views/components/timeline.vue index 2db64d6e0..e0215ad1a 100644 --- a/src/client/app/desktop/views/components/timeline.vue +++ b/src/client/app/desktop/views/components/timeline.vue @@ -1,169 +1,93 @@ diff --git a/src/models/mute.ts b/src/models/mute.ts index e068215c9..8fe4eb2ee 100644 --- a/src/models/mute.ts +++ b/src/models/mute.ts @@ -2,12 +2,12 @@ import * as mongo from 'mongodb'; import db from '../db/mongodb'; const Mute = db.get('mute'); +Mute.createIndex(['muterId', 'muteeId'], { unique: true }); export default Mute; export interface IMute { _id: mongo.ObjectID; createdAt: Date; - deletedAt: Date; muterId: mongo.ObjectID; muteeId: mongo.ObjectID; } diff --git a/src/publishers/stream.ts b/src/publishers/stream.ts index a6d2c2277..2ecbfa0dd 100644 --- a/src/publishers/stream.ts +++ b/src/publishers/stream.ts @@ -45,6 +45,14 @@ class MisskeyEvent { this.publish(`channel-stream:${channelId}`, type, typeof value === 'undefined' ? null : value); } + public publishLocalTimelineStream(note: any): void { + this.redisClient.publish('misskey:local-timeline', JSON.stringify(note)); + } + + public publishGlobalTimelineStream(note: any): void { + this.redisClient.publish('misskey:global-timeline', JSON.stringify(note)); + } + private publish(channel: string, type: string, value?: any): void { const message = value == null ? { type: type } : @@ -58,16 +66,12 @@ const ev = new MisskeyEvent(); export default ev.publishUserStream.bind(ev); +export const publishLocalTimelineStream = ev.publishLocalTimelineStream.bind(ev); +export const publishGlobalTimelineStream = ev.publishGlobalTimelineStream.bind(ev); export const publishDriveStream = ev.publishDriveStream.bind(ev); - export const publishNoteStream = ev.publishNoteStream.bind(ev); - export const publishMessagingStream = ev.publishMessagingStream.bind(ev); - export const publishMessagingIndexStream = ev.publishMessagingIndexStream.bind(ev); - export const publishOthelloStream = ev.publishOthelloStream.bind(ev); - export const publishOthelloGameStream = ev.publishOthelloGameStream.bind(ev); - export const publishChannelStream = ev.publishChannelStream.bind(ev); diff --git a/src/server/api/endpoints.ts b/src/server/api/endpoints.ts index 67f3217fa..e0223c23e 100644 --- a/src/server/api/endpoints.ts +++ b/src/server/api/endpoints.ts @@ -463,6 +463,22 @@ const endpoints: Endpoint[] = [ max: 100 } }, + { + name: 'notes/local-timeline', + withCredential: true, + limit: { + duration: ms('10minutes'), + max: 100 + } + }, + { + name: 'notes/global-timeline', + withCredential: true, + limit: { + duration: ms('10minutes'), + max: 100 + } + }, { name: 'notes/mentions', withCredential: true, diff --git a/src/server/api/endpoints/mute/create.ts b/src/server/api/endpoints/mute/create.ts index 19894d07a..26ae612ca 100644 --- a/src/server/api/endpoints/mute/create.ts +++ b/src/server/api/endpoints/mute/create.ts @@ -30,7 +30,7 @@ module.exports = (params, user) => new Promise(async (res, rej) => { }, { fields: { data: false, - 'profile': false + profile: false } }); @@ -41,8 +41,7 @@ module.exports = (params, user) => new Promise(async (res, rej) => { // Check if already muting const exist = await Mute.findOne({ muterId: muter._id, - muteeId: mutee._id, - deletedAt: { $exists: false } + muteeId: mutee._id }); if (exist !== null) { diff --git a/src/server/api/endpoints/mute/delete.ts b/src/server/api/endpoints/mute/delete.ts index 10096352b..6f617416c 100644 --- a/src/server/api/endpoints/mute/delete.ts +++ b/src/server/api/endpoints/mute/delete.ts @@ -7,10 +7,6 @@ import Mute from '../../../../models/mute'; /** * Unmute a user - * - * @param {any} params - * @param {any} user - * @return {Promise} */ module.exports = (params, user) => new Promise(async (res, rej) => { const muter = user; @@ -30,7 +26,7 @@ module.exports = (params, user) => new Promise(async (res, rej) => { }, { fields: { data: false, - 'profile': false + profile: false } }); @@ -41,8 +37,7 @@ module.exports = (params, user) => new Promise(async (res, rej) => { // Check not muting const exist = await Mute.findOne({ muterId: muter._id, - muteeId: mutee._id, - deletedAt: { $exists: false } + muteeId: mutee._id }); if (exist === null) { @@ -50,12 +45,8 @@ module.exports = (params, user) => new Promise(async (res, rej) => { } // Delete mute - await Mute.update({ + await Mute.remove({ _id: exist._id - }, { - $set: { - deletedAt: new Date() - } }); // Send response diff --git a/src/server/api/endpoints/notes/global-timeline.ts b/src/server/api/endpoints/notes/global-timeline.ts new file mode 100644 index 000000000..07e138ec5 --- /dev/null +++ b/src/server/api/endpoints/notes/global-timeline.ts @@ -0,0 +1,91 @@ +/** + * Module dependencies + */ +import $ from 'cafy'; +import Note from '../../../../models/note'; +import Mute from '../../../../models/mute'; +import { pack } from '../../../../models/note'; + +/** + * Get timeline of global + */ +module.exports = async (params, user, app) => { + // Get 'limit' parameter + const [limit = 10, limitErr] = $(params.limit).optional.number().range(1, 100).$; + if (limitErr) throw 'invalid limit param'; + + // Get 'sinceId' parameter + const [sinceId, sinceIdErr] = $(params.sinceId).optional.id().$; + if (sinceIdErr) throw 'invalid sinceId param'; + + // Get 'untilId' parameter + const [untilId, untilIdErr] = $(params.untilId).optional.id().$; + if (untilIdErr) throw 'invalid untilId param'; + + // Get 'sinceDate' parameter + const [sinceDate, sinceDateErr] = $(params.sinceDate).optional.number().$; + if (sinceDateErr) throw 'invalid sinceDate param'; + + // Get 'untilDate' parameter + const [untilDate, untilDateErr] = $(params.untilDate).optional.number().$; + if (untilDateErr) throw 'invalid untilDate param'; + + // Check if only one of sinceId, untilId, sinceDate, untilDate specified + if ([sinceId, untilId, sinceDate, untilDate].filter(x => x != null).length > 1) { + throw 'only one of sinceId, untilId, sinceDate, untilDate can be specified'; + } + + // ミュートしているユーザーを取得 + const mutedUserIds = (await Mute.find({ + muterId: user._id + })).map(m => m.muteeId); + + //#region Construct query + const sort = { + _id: -1 + }; + + const query = { + // mute + userId: { + $nin: mutedUserIds + }, + '_reply.userId': { + $nin: mutedUserIds + }, + '_renote.userId': { + $nin: mutedUserIds + } + } as any; + + if (sinceId) { + sort._id = 1; + query._id = { + $gt: sinceId + }; + } else if (untilId) { + query._id = { + $lt: untilId + }; + } else if (sinceDate) { + sort._id = 1; + query.createdAt = { + $gt: new Date(sinceDate) + }; + } else if (untilDate) { + query.createdAt = { + $lt: new Date(untilDate) + }; + } + //#endregion + + // Issue query + const timeline = await Note + .find(query, { + limit: limit, + sort: sort + }); + + // Serialize + return await Promise.all(timeline.map(note => pack(note, user))); +}; diff --git a/src/server/api/endpoints/notes/local-timeline.ts b/src/server/api/endpoints/notes/local-timeline.ts new file mode 100644 index 000000000..d63528c3c --- /dev/null +++ b/src/server/api/endpoints/notes/local-timeline.ts @@ -0,0 +1,94 @@ +/** + * Module dependencies + */ +import $ from 'cafy'; +import Note from '../../../../models/note'; +import Mute from '../../../../models/mute'; +import { pack } from '../../../../models/note'; + +/** + * Get timeline of local + */ +module.exports = async (params, user, app) => { + // Get 'limit' parameter + const [limit = 10, limitErr] = $(params.limit).optional.number().range(1, 100).$; + if (limitErr) throw 'invalid limit param'; + + // Get 'sinceId' parameter + const [sinceId, sinceIdErr] = $(params.sinceId).optional.id().$; + if (sinceIdErr) throw 'invalid sinceId param'; + + // Get 'untilId' parameter + const [untilId, untilIdErr] = $(params.untilId).optional.id().$; + if (untilIdErr) throw 'invalid untilId param'; + + // Get 'sinceDate' parameter + const [sinceDate, sinceDateErr] = $(params.sinceDate).optional.number().$; + if (sinceDateErr) throw 'invalid sinceDate param'; + + // Get 'untilDate' parameter + const [untilDate, untilDateErr] = $(params.untilDate).optional.number().$; + if (untilDateErr) throw 'invalid untilDate param'; + + // Check if only one of sinceId, untilId, sinceDate, untilDate specified + if ([sinceId, untilId, sinceDate, untilDate].filter(x => x != null).length > 1) { + throw 'only one of sinceId, untilId, sinceDate, untilDate can be specified'; + } + + // ミュートしているユーザーを取得 + const mutedUserIds = (await Mute.find({ + muterId: user._id + })).map(m => m.muteeId); + + //#region Construct query + const sort = { + _id: -1 + }; + + const query = { + // mute + userId: { + $nin: mutedUserIds + }, + '_reply.userId': { + $nin: mutedUserIds + }, + '_renote.userId': { + $nin: mutedUserIds + }, + + // local + '_user.host': null + } as any; + + if (sinceId) { + sort._id = 1; + query._id = { + $gt: sinceId + }; + } else if (untilId) { + query._id = { + $lt: untilId + }; + } else if (sinceDate) { + sort._id = 1; + query.createdAt = { + $gt: new Date(sinceDate) + }; + } else if (untilDate) { + query.createdAt = { + $lt: new Date(untilDate) + }; + } + //#endregion + + // Issue query + const timeline = await Note + .find(query, { + limit: limit, + sort: sort + }); + + // Serialize + return await Promise.all(timeline.map(note => pack(note, user))); +}; diff --git a/src/server/api/endpoints/notes/timeline.ts b/src/server/api/endpoints/notes/timeline.ts index 5263cfb2a..b5feaac81 100644 --- a/src/server/api/endpoints/notes/timeline.ts +++ b/src/server/api/endpoints/notes/timeline.ts @@ -11,11 +11,6 @@ import { pack } from '../../../../models/note'; /** * Get timeline of myself - * - * @param {any} params - * @param {any} user - * @param {any} app - * @return {Promise} */ module.exports = async (params, user, app) => { // Get 'limit' parameter @@ -56,9 +51,7 @@ module.exports = async (params, user, app) => { // ミュートしているユーザーを取得 mutedUserIds: Mute.find({ - muterId: user._id, - // 削除されたドキュメントは除く - deletedAt: { $exists: false } + muterId: user._id }).then(ms => ms.map(m => m.muteeId)) }); diff --git a/src/server/api/stream/global-timeline.ts b/src/server/api/stream/global-timeline.ts new file mode 100644 index 000000000..f31ce1775 --- /dev/null +++ b/src/server/api/stream/global-timeline.ts @@ -0,0 +1,39 @@ +import * as websocket from 'websocket'; +import * as redis from 'redis'; + +import { IUser } from '../../../models/user'; +import Mute from '../../../models/mute'; + +export default async function( + request: websocket.request, + connection: websocket.connection, + subscriber: redis.RedisClient, + user: IUser +) { + // Subscribe stream + subscriber.subscribe(`misskey:global-timeline`); + + const mute = await Mute.find({ muterId: user._id }); + const mutedUserIds = mute.map(m => m.muteeId.toString()); + + subscriber.on('message', async (_, data) => { + const note = JSON.parse(data); + + //#region 流れてきたNoteがミュートしているユーザーが関わるものだったら無視する + if (mutedUserIds.indexOf(note.userId) != -1) { + return; + } + if (note.reply != null && mutedUserIds.indexOf(note.reply.userId) != -1) { + return; + } + if (note.renote != null && mutedUserIds.indexOf(note.renote.userId) != -1) { + return; + } + //#endregion + + connection.send(JSON.stringify({ + type: 'note', + body: note + })); + }); +} diff --git a/src/server/api/stream/home.ts b/src/server/api/stream/home.ts index e9c0924f3..a9d6ff241 100644 --- a/src/server/api/stream/home.ts +++ b/src/server/api/stream/home.ts @@ -21,10 +21,7 @@ export default async function( // Subscribe Home stream channel subscriber.subscribe(`misskey:user-stream:${user._id}`); - const mute = await Mute.find({ - muterId: user._id, - deletedAt: { $exists: false } - }); + const mute = await Mute.find({ muterId: user._id }); const mutedUserIds = mute.map(m => m.muteeId.toString()); subscriber.on('message', async (channel, data) => { @@ -33,6 +30,7 @@ export default async function( try { const x = JSON.parse(data); + //#region 流れてきたメッセージがミュートしているユーザーが関わるものだったら無視する if (x.type == 'note') { if (mutedUserIds.indexOf(x.body.userId) != -1) { return; @@ -48,6 +46,7 @@ export default async function( return; } } + //#endregion connection.send(data); } catch (e) { diff --git a/src/server/api/stream/local-timeline.ts b/src/server/api/stream/local-timeline.ts new file mode 100644 index 000000000..a790ba878 --- /dev/null +++ b/src/server/api/stream/local-timeline.ts @@ -0,0 +1,39 @@ +import * as websocket from 'websocket'; +import * as redis from 'redis'; + +import { IUser } from '../../../models/user'; +import Mute from '../../../models/mute'; + +export default async function( + request: websocket.request, + connection: websocket.connection, + subscriber: redis.RedisClient, + user: IUser +) { + // Subscribe stream + subscriber.subscribe(`misskey:local-timeline`); + + const mute = await Mute.find({ muterId: user._id }); + const mutedUserIds = mute.map(m => m.muteeId.toString()); + + subscriber.on('message', async (_, data) => { + const note = JSON.parse(data); + + //#region 流れてきたNoteがミュートしているユーザーが関わるものだったら無視する + if (mutedUserIds.indexOf(note.userId) != -1) { + return; + } + if (note.reply != null && mutedUserIds.indexOf(note.reply.userId) != -1) { + return; + } + if (note.renote != null && mutedUserIds.indexOf(note.renote.userId) != -1) { + return; + } + //#endregion + + connection.send(JSON.stringify({ + type: 'note', + body: note + })); + }); +} diff --git a/src/server/api/streaming.ts b/src/server/api/streaming.ts index d586d7c08..ce1325364 100644 --- a/src/server/api/streaming.ts +++ b/src/server/api/streaming.ts @@ -4,6 +4,8 @@ import * as redis from 'redis'; import config from '../../config'; import homeStream from './stream/home'; +import localTimelineStream from './stream/local-timeline'; +import globalTimelineStream from './stream/global-timeline'; import driveStream from './stream/drive'; import messagingStream from './stream/messaging'; import messagingIndexStream from './stream/messaging-index'; @@ -64,8 +66,10 @@ module.exports = (server: http.Server) => { return; } - const channel = + const channel: any = request.resourceURL.pathname === '/' ? homeStream : + request.resourceURL.pathname === '/local-timeline' ? localTimelineStream : + request.resourceURL.pathname === '/global-timeline' ? globalTimelineStream : request.resourceURL.pathname === '/drive' ? driveStream : request.resourceURL.pathname === '/messaging' ? messagingStream : request.resourceURL.pathname === '/messaging-index' ? messagingIndexStream : diff --git a/src/services/note/create.ts b/src/services/note/create.ts index 781337daf..e35e5ecfb 100644 --- a/src/services/note/create.ts +++ b/src/services/note/create.ts @@ -1,6 +1,6 @@ import Note, { pack, INote } from '../../models/note'; import User, { isLocalUser, IUser, isRemoteUser } from '../../models/user'; -import stream from '../../publishers/stream'; +import stream, { publishLocalTimelineStream, publishGlobalTimelineStream } from '../../publishers/stream'; import Following from '../../models/following'; import { deliver } from '../../queue'; import renderNote from '../../remote/activitypub/renderer/note'; @@ -105,11 +105,17 @@ export default async (user: IUser, data: { // タイムラインへの投稿 if (note.channelId == null) { - // Publish event to myself's stream if (isLocalUser(user)) { + // Publish event to myself's stream stream(note.userId, 'note', noteObj); + + // Publish note to local timeline stream + publishLocalTimelineStream(noteObj); } + // Publish note to global timeline stream + publishGlobalTimelineStream(noteObj); + // Fetch all followers const followers = await Following.aggregate([{ $lookup: {