From ad38cd2605a7cd857618a5498651d5b10e7b1bfc Mon Sep 17 00:00:00 2001
From: Akihiko Odaki <nekomanma@pixiv.co.jp>
Date: Mon, 2 Apr 2018 13:33:46 +0900
Subject: [PATCH] Introduce publishers directory

---
 src/drive/add-file.ts                             |  2 +-
 src/processor/http/follow.ts                      |  4 ++--
 src/{ => publishers}/notify.ts                    | 12 ++++++------
 src/{ => publishers}/push-sw.ts                   |  4 ++--
 src/{event.ts => publishers/stream.ts}            |  9 +--------
 src/server/api/common/read-messaging-message.ts   |  6 +++---
 src/server/api/common/read-notification.ts        |  2 +-
 src/server/api/endpoints/drive/files/update.ts    |  2 +-
 src/server/api/endpoints/drive/folders/create.ts  |  2 +-
 src/server/api/endpoints/drive/folders/update.ts  |  2 +-
 src/server/api/endpoints/following/delete.ts      |  2 +-
 src/server/api/endpoints/i/regenerate_token.ts    |  2 +-
 src/server/api/endpoints/i/update.ts              |  2 +-
 .../api/endpoints/i/update_client_setting.ts      |  2 +-
 src/server/api/endpoints/i/update_home.ts         |  2 +-
 src/server/api/endpoints/i/update_mobile_home.ts  |  2 +-
 .../api/endpoints/messaging/messages/create.ts    |  5 +++--
 .../endpoints/notifications/mark_as_read_all.ts   |  2 +-
 src/server/api/endpoints/othello/match.ts         |  2 +-
 src/server/api/endpoints/posts/create.ts          | 15 ++++++++-------
 src/server/api/endpoints/posts/polls/vote.ts      |  4 ++--
 .../api/endpoints/posts/reactions/create.ts       |  5 +++--
 .../api/endpoints/posts/reactions/delete.ts       |  2 +-
 src/server/api/private/signin.ts                  |  2 +-
 src/server/api/service/twitter.ts                 |  2 +-
 src/server/api/stream/othello-game.ts             |  2 +-
 src/server/api/stream/othello.ts                  |  2 +-
 27 files changed, 48 insertions(+), 52 deletions(-)
 rename src/{ => publishers}/notify.ts (81%)
 rename src/{ => publishers}/push-sw.ts (92%)
 rename src/{event.ts => publishers/stream.ts} (92%)

diff --git a/src/drive/add-file.ts b/src/drive/add-file.ts
index db13e04be..4a718a9da 100644
--- a/src/drive/add-file.ts
+++ b/src/drive/add-file.ts
@@ -13,7 +13,7 @@ import prominence = require('prominence');
 import DriveFile, { getGridFSBucket } from '../models/drive-file';
 import DriveFolder from '../models/drive-folder';
 import { pack } from '../models/drive-file';
-import event, { publishDriveStream } from '../event';
+import event, { publishDriveStream } from '../publishers/stream';
 import getAcct from '../user/get-acct';
 import config from '../config';
 
diff --git a/src/processor/http/follow.ts b/src/processor/http/follow.ts
index d6ce00006..a7e4fa23d 100644
--- a/src/processor/http/follow.ts
+++ b/src/processor/http/follow.ts
@@ -3,8 +3,8 @@ import { sign } from 'http-signature';
 import { URL } from 'url';
 import User, { isLocalUser, pack as packUser, ILocalUser } from '../../models/user';
 import Following from '../../models/following';
-import event from '../../event';
-import notify from '../../notify';
+import event from '../../publishers/stream';
+import notify from '../../publishers/notify';
 import context from '../../remote/activitypub/renderer/context';
 import render from '../../remote/activitypub/renderer/follow';
 import config from '../../config';
diff --git a/src/notify.ts b/src/publishers/notify.ts
similarity index 81%
rename from src/notify.ts
rename to src/publishers/notify.ts
index 228317b88..2b89515d4 100644
--- a/src/notify.ts
+++ b/src/publishers/notify.ts
@@ -1,8 +1,8 @@
 import * as mongo from 'mongodb';
-import Notification from './models/notification';
-import Mute from './models/mute';
-import event from './event';
-import { pack } from './models/notification';
+import Notification from '../models/notification';
+import Mute from '../models/mute';
+import { pack } from '../models/notification';
+import stream from './stream';
 
 export default (
 	notifiee: mongo.ObjectID,
@@ -26,7 +26,7 @@ export default (
 	resolve(notification);
 
 	// Publish notification event
-	event(notifiee, 'notification',
+	stream(notifiee, 'notification',
 		await pack(notification));
 
 	// 3秒経っても(今回作成した)通知が既読にならなかったら「未読の通知がありますよ」イベントを発行する
@@ -44,7 +44,7 @@ export default (
 			}
 			//#endregion
 
-			event(notifiee, 'unread_notification', await pack(notification));
+			stream(notifiee, 'unread_notification', await pack(notification));
 		}
 	}, 3000);
 });
diff --git a/src/push-sw.ts b/src/publishers/push-sw.ts
similarity index 92%
rename from src/push-sw.ts
rename to src/publishers/push-sw.ts
index fcef7796d..aab91df62 100644
--- a/src/push-sw.ts
+++ b/src/publishers/push-sw.ts
@@ -1,7 +1,7 @@
 const push = require('web-push');
 import * as mongo from 'mongodb';
-import Subscription from './models/sw-subscription';
-import config from './config';
+import Subscription from '../models/sw-subscription';
+import config from '../config';
 
 if (config.sw) {
 	// アプリケーションの連絡先と、サーバーサイドの鍵ペアの情報を登録
diff --git a/src/event.ts b/src/publishers/stream.ts
similarity index 92%
rename from src/event.ts
rename to src/publishers/stream.ts
index 81876b3cf..498ff33f3 100644
--- a/src/event.ts
+++ b/src/publishers/stream.ts
@@ -1,7 +1,6 @@
 import * as mongo from 'mongodb';
 import * as redis from 'redis';
-import swPush from './push-sw';
-import config from './config';
+import config from '../config';
 
 type ID = string | mongo.ObjectID;
 
@@ -18,10 +17,6 @@ class MisskeyEvent {
 		this.publish(`user-stream:${userId}`, type, typeof value === 'undefined' ? null : value);
 	}
 
-	public publishSw(userId: ID, type: string, value?: any): void {
-		swPush(userId, type, value);
-	}
-
 	public publishDriveStream(userId: ID, type: string, value?: any): void {
 		this.publish(`drive-stream:${userId}`, type, typeof value === 'undefined' ? null : value);
 	}
@@ -63,8 +58,6 @@ const ev = new MisskeyEvent();
 
 export default ev.publishUserStream.bind(ev);
 
-export const pushSw = ev.publishSw.bind(ev);
-
 export const publishDriveStream = ev.publishDriveStream.bind(ev);
 
 export const publishPostStream = ev.publishPostStream.bind(ev);
diff --git a/src/server/api/common/read-messaging-message.ts b/src/server/api/common/read-messaging-message.ts
index 755cf1f50..c52f9363b 100644
--- a/src/server/api/common/read-messaging-message.ts
+++ b/src/server/api/common/read-messaging-message.ts
@@ -1,9 +1,9 @@
 import * as mongo from 'mongodb';
 import Message from '../../../models/messaging-message';
 import { IMessagingMessage as IMessage } from '../../../models/messaging-message';
-import publishUserStream from '../../../event';
-import { publishMessagingStream } from '../../../event';
-import { publishMessagingIndexStream } from '../../../event';
+import publishUserStream from '../../../publishers/stream';
+import { publishMessagingStream } from '../../../publishers/stream';
+import { publishMessagingIndexStream } from '../../../publishers/stream';
 
 /**
  * Mark as read message(s)
diff --git a/src/server/api/common/read-notification.ts b/src/server/api/common/read-notification.ts
index b51c0ca00..9bd41519f 100644
--- a/src/server/api/common/read-notification.ts
+++ b/src/server/api/common/read-notification.ts
@@ -1,6 +1,6 @@
 import * as mongo from 'mongodb';
 import { default as Notification, INotification } from '../../../models/notification';
-import publishUserStream from '../../../event';
+import publishUserStream from '../../../publishers/stream';
 
 /**
  * Mark as read notification(s)
diff --git a/src/server/api/endpoints/drive/files/update.ts b/src/server/api/endpoints/drive/files/update.ts
index 85bd2110f..c783ad8b3 100644
--- a/src/server/api/endpoints/drive/files/update.ts
+++ b/src/server/api/endpoints/drive/files/update.ts
@@ -4,7 +4,7 @@
 import $ from 'cafy';
 import DriveFolder from '../../../../../models/drive-folder';
 import DriveFile, { validateFileName, pack } from '../../../../../models/drive-file';
-import { publishDriveStream } from '../../../../../event';
+import { publishDriveStream } from '../../../../../publishers/stream';
 
 /**
  * Update a file
diff --git a/src/server/api/endpoints/drive/folders/create.ts b/src/server/api/endpoints/drive/folders/create.ts
index d9d39a918..f34d0019d 100644
--- a/src/server/api/endpoints/drive/folders/create.ts
+++ b/src/server/api/endpoints/drive/folders/create.ts
@@ -3,7 +3,7 @@
  */
 import $ from 'cafy';
 import DriveFolder, { isValidFolderName, pack } from '../../../../../models/drive-folder';
-import { publishDriveStream } from '../../../../../event';
+import { publishDriveStream } from '../../../../../publishers/stream';
 
 /**
  * Create drive folder
diff --git a/src/server/api/endpoints/drive/folders/update.ts b/src/server/api/endpoints/drive/folders/update.ts
index 1cea05d71..dd7e8f5c8 100644
--- a/src/server/api/endpoints/drive/folders/update.ts
+++ b/src/server/api/endpoints/drive/folders/update.ts
@@ -3,7 +3,7 @@
  */
 import $ from 'cafy';
 import DriveFolder, { isValidFolderName, pack } from '../../../../../models/drive-folder';
-import { publishDriveStream } from '../../../../../event';
+import { publishDriveStream } from '../../../../../publishers/stream';
 
 /**
  * Update a folder
diff --git a/src/server/api/endpoints/following/delete.ts b/src/server/api/endpoints/following/delete.ts
index 77a6cebee..3facfdcdd 100644
--- a/src/server/api/endpoints/following/delete.ts
+++ b/src/server/api/endpoints/following/delete.ts
@@ -4,7 +4,7 @@
 import $ from 'cafy';
 import User, { pack as packUser } from '../../../../models/user';
 import Following from '../../../../models/following';
-import event from '../../../../event';
+import event from '../../../../publishers/stream';
 
 /**
  * Unfollow a user
diff --git a/src/server/api/endpoints/i/regenerate_token.ts b/src/server/api/endpoints/i/regenerate_token.ts
index d7cb69784..9aa6725f8 100644
--- a/src/server/api/endpoints/i/regenerate_token.ts
+++ b/src/server/api/endpoints/i/regenerate_token.ts
@@ -4,7 +4,7 @@
 import $ from 'cafy';
 import * as bcrypt from 'bcryptjs';
 import User from '../../../../models/user';
-import event from '../../../../event';
+import event from '../../../../publishers/stream';
 import generateUserToken from '../../common/generate-native-user-token';
 
 /**
diff --git a/src/server/api/endpoints/i/update.ts b/src/server/api/endpoints/i/update.ts
index c4ec41339..279b062f5 100644
--- a/src/server/api/endpoints/i/update.ts
+++ b/src/server/api/endpoints/i/update.ts
@@ -3,7 +3,7 @@
  */
 import $ from 'cafy';
 import User, { isValidName, isValidDescription, isValidLocation, isValidBirthday, pack } from '../../../../models/user';
-import event from '../../../../event';
+import event from '../../../../publishers/stream';
 import config from '../../../../config';
 
 /**
diff --git a/src/server/api/endpoints/i/update_client_setting.ts b/src/server/api/endpoints/i/update_client_setting.ts
index 263ac6d07..10741aceb 100644
--- a/src/server/api/endpoints/i/update_client_setting.ts
+++ b/src/server/api/endpoints/i/update_client_setting.ts
@@ -3,7 +3,7 @@
  */
 import $ from 'cafy';
 import User, { pack } from '../../../../models/user';
-import event from '../../../../event';
+import event from '../../../../publishers/stream';
 
 /**
  * Update myself
diff --git a/src/server/api/endpoints/i/update_home.ts b/src/server/api/endpoints/i/update_home.ts
index 227b8dd5a..91be0714d 100644
--- a/src/server/api/endpoints/i/update_home.ts
+++ b/src/server/api/endpoints/i/update_home.ts
@@ -3,7 +3,7 @@
  */
 import $ from 'cafy';
 import User from '../../../../models/user';
-import event from '../../../../event';
+import event from '../../../../publishers/stream';
 
 module.exports = async (params, user) => new Promise(async (res, rej) => {
 	// Get 'home' parameter
diff --git a/src/server/api/endpoints/i/update_mobile_home.ts b/src/server/api/endpoints/i/update_mobile_home.ts
index 007eb2eab..1efda120d 100644
--- a/src/server/api/endpoints/i/update_mobile_home.ts
+++ b/src/server/api/endpoints/i/update_mobile_home.ts
@@ -3,7 +3,7 @@
  */
 import $ from 'cafy';
 import User from '../../../../models/user';
-import event from '../../../../event';
+import event from '../../../../publishers/stream';
 
 module.exports = async (params, user) => new Promise(async (res, rej) => {
 	// Get 'home' parameter
diff --git a/src/server/api/endpoints/messaging/messages/create.ts b/src/server/api/endpoints/messaging/messages/create.ts
index 604da32d2..085e75e6c 100644
--- a/src/server/api/endpoints/messaging/messages/create.ts
+++ b/src/server/api/endpoints/messaging/messages/create.ts
@@ -9,8 +9,9 @@ import User from '../../../../../models/user';
 import Mute from '../../../../../models/mute';
 import DriveFile from '../../../../../models/drive-file';
 import { pack } from '../../../../../models/messaging-message';
-import publishUserStream from '../../../../../event';
-import { publishMessagingStream, publishMessagingIndexStream, pushSw } from '../../../../../event';
+import publishUserStream from '../../../../../publishers/stream';
+import { publishMessagingStream, publishMessagingIndexStream } from '../../../../../publishers/stream';
+import pushSw from '../../../../../publishers/push-sw';
 import html from '../../../../../text/html';
 import parse from '../../../../../text/parse';
 import config from '../../../../../config';
diff --git a/src/server/api/endpoints/notifications/mark_as_read_all.ts b/src/server/api/endpoints/notifications/mark_as_read_all.ts
index 3ba00a907..01c914583 100644
--- a/src/server/api/endpoints/notifications/mark_as_read_all.ts
+++ b/src/server/api/endpoints/notifications/mark_as_read_all.ts
@@ -2,7 +2,7 @@
  * Module dependencies
  */
 import Notification from '../../../../models/notification';
-import event from '../../../../event';
+import event from '../../../../publishers/stream';
 
 /**
  * Mark as read all notifications
diff --git a/src/server/api/endpoints/othello/match.ts b/src/server/api/endpoints/othello/match.ts
index 1fc46ab90..d9274f8f9 100644
--- a/src/server/api/endpoints/othello/match.ts
+++ b/src/server/api/endpoints/othello/match.ts
@@ -2,7 +2,7 @@ import $ from 'cafy';
 import Matching, { pack as packMatching } from '../../../../models/othello-matching';
 import OthelloGame, { pack as packGame } from '../../../../models/othello-game';
 import User from '../../../../models/user';
-import publishUserStream, { publishOthelloStream } from '../../../../event';
+import publishUserStream, { publishOthelloStream } from '../../../../publishers/stream';
 import { eighteight } from '../../../../othello/maps';
 
 module.exports = (params, user) => new Promise(async (res, rej) => {
diff --git a/src/server/api/endpoints/posts/create.ts b/src/server/api/endpoints/posts/create.ts
index bf08fe283..34a3aa190 100644
--- a/src/server/api/endpoints/posts/create.ts
+++ b/src/server/api/endpoints/posts/create.ts
@@ -15,8 +15,9 @@ import Watching from '../../../../models/post-watching';
 import ChannelWatching from '../../../../models/channel-watching';
 import { pack } from '../../../../models/post';
 import watch from '../../common/watch-post';
-import event, { pushSw, publishChannelStream } from '../../../../event';
-import notify from '../../../../notify';
+import stream, { publishChannelStream } from '../../../../publishers/stream';
+import notify from '../../../../publishers/notify';
+import pushSw from '../../../../publishers/push-sw';
 import getAcct from '../../../../user/get-acct';
 import parseAcct from '../../../../user/parse-acct';
 import config from '../../../../config';
@@ -306,7 +307,7 @@ module.exports = (params, user: ILocalUser, app) => new Promise(async (res, rej)
 			});
 			const mentioneesMutedUserIds = mentioneeMutes.map(m => m.muteeId.toString());
 			if (mentioneesMutedUserIds.indexOf(user._id.toString()) == -1) {
-				event(mentionee, reason, postObj);
+				stream(mentionee, reason, postObj);
 				pushSw(mentionee, reason, postObj);
 			}
 		}
@@ -315,7 +316,7 @@ module.exports = (params, user: ILocalUser, app) => new Promise(async (res, rej)
 	// タイムラインへの投稿
 	if (!channel) {
 		// Publish event to myself's stream
-		event(user._id, 'post', postObj);
+		stream(user._id, 'post', postObj);
 
 		// Fetch all followers
 		const followers = await Following
@@ -330,7 +331,7 @@ module.exports = (params, user: ILocalUser, app) => new Promise(async (res, rej)
 
 		// Publish event to followers stream
 		followers.forEach(following =>
-			event(following.followerId, 'post', postObj));
+			stream(following.followerId, 'post', postObj));
 	}
 
 	// チャンネルへの投稿
@@ -354,7 +355,7 @@ module.exports = (params, user: ILocalUser, app) => new Promise(async (res, rej)
 
 		// チャンネルの視聴者(のタイムライン)に配信
 		watches.forEach(w => {
-			event(w.userId, 'post', postObj);
+			stream(w.userId, 'post', postObj);
 		});
 	}
 
@@ -448,7 +449,7 @@ module.exports = (params, user: ILocalUser, app) => new Promise(async (res, rej)
 		} else {
 			// Publish event
 			if (!user._id.equals(repost.userId)) {
-				event(repost.userId, 'repost', postObj);
+				stream(repost.userId, 'repost', postObj);
 			}
 		}
 
diff --git a/src/server/api/endpoints/posts/polls/vote.ts b/src/server/api/endpoints/posts/polls/vote.ts
index be1fd7b5d..029fb9323 100644
--- a/src/server/api/endpoints/posts/polls/vote.ts
+++ b/src/server/api/endpoints/posts/polls/vote.ts
@@ -6,8 +6,8 @@ import Vote from '../../../../../models/poll-vote';
 import Post from '../../../../../models/post';
 import Watching from '../../../../../models/post-watching';
 import watch from '../../../common/watch-post';
-import { publishPostStream } from '../../../../../event';
-import notify from '../../../../../notify';
+import { publishPostStream } from '../../../../../publishers/stream';
+import notify from '../../../../../publishers/notify';
 
 /**
  * Vote poll of a post
diff --git a/src/server/api/endpoints/posts/reactions/create.ts b/src/server/api/endpoints/posts/reactions/create.ts
index 408b2483a..8b5f1e57d 100644
--- a/src/server/api/endpoints/posts/reactions/create.ts
+++ b/src/server/api/endpoints/posts/reactions/create.ts
@@ -7,8 +7,9 @@ import Post, { pack as packPost } from '../../../../../models/post';
 import { pack as packUser } from '../../../../../models/user';
 import Watching from '../../../../../models/post-watching';
 import watch from '../../../common/watch-post';
-import { publishPostStream, pushSw } from '../../../../../event';
-import notify from '../../../../../notify';
+import { publishPostStream } from '../../../../../publishers/stream';
+import notify from '../../../../../publishers/notify';
+import pushSw from '../../../../../publishers/push-sw';
 
 /**
  * React to a post
diff --git a/src/server/api/endpoints/posts/reactions/delete.ts b/src/server/api/endpoints/posts/reactions/delete.ts
index 11f5c7daf..3a88bbd7c 100644
--- a/src/server/api/endpoints/posts/reactions/delete.ts
+++ b/src/server/api/endpoints/posts/reactions/delete.ts
@@ -4,7 +4,7 @@
 import $ from 'cafy';
 import Reaction from '../../../../../models/post-reaction';
 import Post from '../../../../../models/post';
-// import event from '../../../event';
+// import event from '../../../publishers/stream';
 
 /**
  * Unreact to a post
diff --git a/src/server/api/private/signin.ts b/src/server/api/private/signin.ts
index bf883ee27..e0bd67d1c 100644
--- a/src/server/api/private/signin.ts
+++ b/src/server/api/private/signin.ts
@@ -3,7 +3,7 @@ import * as bcrypt from 'bcryptjs';
 import * as speakeasy from 'speakeasy';
 import User, { ILocalUser } from '../../../models/user';
 import Signin, { pack } from '../../../models/signin';
-import event from '../../../event';
+import event from '../../../publishers/stream';
 import signin from '../common/signin';
 import config from '../../../config';
 
diff --git a/src/server/api/service/twitter.ts b/src/server/api/service/twitter.ts
index 69fa5f3c6..77b932b13 100644
--- a/src/server/api/service/twitter.ts
+++ b/src/server/api/service/twitter.ts
@@ -6,7 +6,7 @@ import * as uuid from 'uuid';
 import autwh from 'autwh';
 import redis from '../../../db/redis';
 import User, { pack } from '../../../models/user';
-import event from '../../../event';
+import event from '../../../publishers/stream';
 import config from '../../../config';
 import signin from '../common/signin';
 
diff --git a/src/server/api/stream/othello-game.ts b/src/server/api/stream/othello-game.ts
index 6eb610331..841e54261 100644
--- a/src/server/api/stream/othello-game.ts
+++ b/src/server/api/stream/othello-game.ts
@@ -2,7 +2,7 @@ import * as websocket from 'websocket';
 import * as redis from 'redis';
 import * as CRC32 from 'crc-32';
 import OthelloGame, { pack } from '../../../models/othello-game';
-import { publishOthelloGameStream } from '../../../event';
+import { publishOthelloGameStream } from '../../../publishers/stream';
 import Othello from '../../../othello/core';
 import * as maps from '../../../othello/maps';
 import { ParsedUrlQuery } from 'querystring';
diff --git a/src/server/api/stream/othello.ts b/src/server/api/stream/othello.ts
index 4c292056d..fa62b0583 100644
--- a/src/server/api/stream/othello.ts
+++ b/src/server/api/stream/othello.ts
@@ -2,7 +2,7 @@ import * as mongo from 'mongodb';
 import * as websocket from 'websocket';
 import * as redis from 'redis';
 import Matching, { pack } from '../../../models/othello-matching';
-import publishUserStream from '../../../event';
+import publishUserStream from '../../../publishers/stream';
 
 export default function(request: websocket.request, connection: websocket.connection, subscriber: redis.RedisClient, user: any): void {
 	// Subscribe othello stream