From b6aeacdeb942beb7b5b2f6ac8cf4a89163e59153 Mon Sep 17 00:00:00 2001 From: syuilo Date: Fri, 6 Apr 2018 03:10:25 +0900 Subject: RENAME: api --> services --- src/services/drive/add-file.ts | 314 +++++++++++++++++++++++++++++++ src/services/drive/upload-from-url.ts | 52 +++++ src/services/following/create.ts | 72 +++++++ src/services/following/delete.ts | 64 +++++++ src/services/post/create.ts | 344 ++++++++++++++++++++++++++++++++++ src/services/post/reaction/create.ts | 0 src/services/post/watch.ts | 26 +++ 7 files changed, 872 insertions(+) create mode 100644 src/services/drive/add-file.ts create mode 100644 src/services/drive/upload-from-url.ts create mode 100644 src/services/following/create.ts create mode 100644 src/services/following/delete.ts create mode 100644 src/services/post/create.ts create mode 100644 src/services/post/reaction/create.ts create mode 100644 src/services/post/watch.ts (limited to 'src/services') diff --git a/src/services/drive/add-file.ts b/src/services/drive/add-file.ts new file mode 100644 index 0000000000..64a2f18340 --- /dev/null +++ b/src/services/drive/add-file.ts @@ -0,0 +1,314 @@ +import { Buffer } from 'buffer'; +import * as fs from 'fs'; +import * as tmp from 'tmp'; +import * as stream from 'stream'; + +import * as mongodb from 'mongodb'; +import * as crypto from 'crypto'; +import * as _gm from 'gm'; +import * as debug from 'debug'; +import fileType = require('file-type'); +import prominence = require('prominence'); + +import DriveFile, { IMetadata, getGridFSBucket } from '../../models/drive-file'; +import DriveFolder from '../../models/drive-folder'; +import { pack } from '../../models/drive-file'; +import event, { publishDriveStream } from '../../publishers/stream'; +import getAcct from '../../acct/render'; +import config from '../../config'; + +const gm = _gm.subClass({ + imageMagick: true +}); + +const log = debug('misskey:drive:add-file'); + +const tmpFile = (): Promise => new Promise((resolve, reject) => { + tmp.file((e, path) => { + if (e) return reject(e); + resolve(path); + }); +}); + +const addToGridFS = (name: string, readable: stream.Readable, type: string, metadata: any): Promise => + getGridFSBucket() + .then(bucket => new Promise((resolve, reject) => { + const writeStream = bucket.openUploadStream(name, { contentType: type, metadata }); + writeStream.once('finish', (doc) => { resolve(doc); }); + writeStream.on('error', reject); + readable.pipe(writeStream); + })); + +const addFile = async ( + user: any, + path: string, + name: string = null, + comment: string = null, + folderId: mongodb.ObjectID = null, + force: boolean = false, + uri: string = null +) => { + log(`registering ${name} (user: ${getAcct(user)}, path: ${path})`); + + // Calculate hash, get content type and get file size + const [hash, [mime, ext], size] = await Promise.all([ + // hash + ((): Promise => new Promise((res, rej) => { + const readable = fs.createReadStream(path); + const hash = crypto.createHash('md5'); + const chunks = []; + readable + .on('error', rej) + .pipe(hash) + .on('error', rej) + .on('data', (chunk) => chunks.push(chunk)) + .on('end', () => { + const buffer = Buffer.concat(chunks); + res(buffer.toString('hex')); + }); + }))(), + // mime + ((): Promise<[string, string | null]> => new Promise((res, rej) => { + const readable = fs.createReadStream(path); + readable + .on('error', rej) + .once('data', (buffer: Buffer) => { + readable.destroy(); + const type = fileType(buffer); + if (type) { + return res([type.mime, type.ext]); + } else { + // 種類が同定できなかったら application/octet-stream にする + return res(['application/octet-stream', null]); + } + }); + }))(), + // size + ((): Promise => new Promise((res, rej) => { + fs.stat(path, (err, stats) => { + if (err) return rej(err); + res(stats.size); + }); + }))() + ]); + + log(`hash: ${hash}, mime: ${mime}, ext: ${ext}, size: ${size}`); + + // detect name + const detectedName: string = name || (ext ? `untitled.${ext}` : 'untitled'); + + if (!force) { + // Check if there is a file with the same hash + const much = await DriveFile.findOne({ + md5: hash, + 'metadata.userId': user._id + }); + + if (much !== null) { + log('file with same hash is found'); + return much; + } else { + log('file with same hash is not found'); + } + } + + const [wh, averageColor, folder] = await Promise.all([ + // Width and height (when image) + (async () => { + // 画像かどうか + if (!/^image\/.*$/.test(mime)) { + return null; + } + + const imageType = mime.split('/')[1]; + + // 画像でもPNGかJPEGかGIFでないならスキップ + if (imageType != 'png' && imageType != 'jpeg' && imageType != 'gif') { + return null; + } + + log('calculate image width and height...'); + + // Calculate width and height + const g = gm(fs.createReadStream(path), name); + const size = await prominence(g).size(); + + log(`image width and height is calculated: ${size.width}, ${size.height}`); + + return [size.width, size.height]; + })(), + // average color (when image) + (async () => { + // 画像かどうか + if (!/^image\/.*$/.test(mime)) { + return null; + } + + const imageType = mime.split('/')[1]; + + // 画像でもPNGかJPEGでないならスキップ + if (imageType != 'png' && imageType != 'jpeg') { + return null; + } + + log('calculate average color...'); + + const buffer = await prominence(gm(fs.createReadStream(path), name) + .setFormat('ppm') + .resize(1, 1)) // 1pxのサイズに縮小して平均色を取得するというハック + .toBuffer(); + + const r = buffer.readUInt8(buffer.length - 3); + const g = buffer.readUInt8(buffer.length - 2); + const b = buffer.readUInt8(buffer.length - 1); + + log(`average color is calculated: ${r}, ${g}, ${b}`); + + return [r, g, b]; + })(), + // folder + (async () => { + if (!folderId) { + return null; + } + const driveFolder = await DriveFolder.findOne({ + _id: folderId, + userId: user._id + }); + if (!driveFolder) { + throw 'folder-not-found'; + } + return driveFolder; + })(), + // usage checker + (async () => { + // Calculate drive usage + const usage = await DriveFile + .aggregate([{ + $match: { 'metadata.userId': user._id } + }, { + $project: { + length: true + } + }, { + $group: { + _id: null, + usage: { $sum: '$length' } + } + }]) + .then((aggregates: any[]) => { + if (aggregates.length > 0) { + return aggregates[0].usage; + } + return 0; + }); + + log(`drive usage is ${usage}`); + + // If usage limit exceeded + if (usage + size > user.driveCapacity) { + throw 'no-free-space'; + } + })() + ]); + + const readable = fs.createReadStream(path); + + const properties = {}; + + if (wh) { + properties['width'] = wh[0]; + properties['height'] = wh[1]; + } + + if (averageColor) { + properties['avgColor'] = averageColor; + } + + const metadata = { + userId: user._id, + folderId: folder !== null ? folder._id : null, + comment: comment, + properties: properties + } as IMetadata; + + if (uri !== null) { + metadata.uri = uri; + } + + return addToGridFS(detectedName, readable, mime, metadata); +}; + +/** + * Add file to drive + * + * @param user User who wish to add file + * @param file File path or readableStream + * @param comment Comment + * @param type File type + * @param folderId Folder ID + * @param force If set to true, forcibly upload the file even if there is a file with the same hash. + * @return Object that represents added file + */ +export default (user: any, file: string | stream.Readable, ...args) => new Promise((resolve, reject) => { + // Get file path + new Promise((res: (v: [string, boolean]) => void, rej) => { + if (typeof file === 'string') { + res([file, false]); + return; + } + if (typeof file === 'object' && typeof file.read === 'function') { + tmpFile() + .then(path => { + const readable: stream.Readable = file; + const writable = fs.createWriteStream(path); + readable + .on('error', rej) + .on('end', () => { + res([path, true]); + }) + .pipe(writable) + .on('error', rej); + }) + .catch(rej); + } + rej(new Error('un-compatible file.')); + }) + .then(([path, shouldCleanup]): Promise => new Promise((res, rej) => { + addFile(user, path, ...args) + .then(file => { + res(file); + if (shouldCleanup) { + fs.unlink(path, (e) => { + if (e) log(e.stack); + }); + } + }) + .catch(rej); + })) + .then(file => { + log(`drive file has been created ${file._id}`); + resolve(file); + + pack(file).then(serializedFile => { + // Publish drive_file_created event + event(user._id, 'drive_file_created', serializedFile); + publishDriveStream(user._id, 'file_created', serializedFile); + + // Register to search database + if (config.elasticsearch.enable) { + const es = require('../db/elasticsearch'); + es.index({ + index: 'misskey', + type: 'drive_file', + id: file._id.toString(), + body: { + name: file.name, + userId: user._id.toString() + } + }); + } + }); + }) + .catch(reject); +}); diff --git a/src/services/drive/upload-from-url.ts b/src/services/drive/upload-from-url.ts new file mode 100644 index 0000000000..676586cd15 --- /dev/null +++ b/src/services/drive/upload-from-url.ts @@ -0,0 +1,52 @@ +import * as URL from 'url'; +import { IDriveFile, validateFileName } from '../../models/drive-file'; +import create from './add-file'; +import * as debug from 'debug'; +import * as tmp from 'tmp'; +import * as fs from 'fs'; +import * as request from 'request'; + +const log = debug('misskey:drive:upload-from-url'); + +export default async (url, user, folderId = null, uri = null): Promise => { + log(`REQUESTED: ${url}`); + + let name = URL.parse(url).pathname.split('/').pop(); + if (!validateFileName(name)) { + name = null; + } + + log(`name: ${name}`); + + // Create temp file + const path = await new Promise((res: (string) => void, rej) => { + tmp.file((e, path) => { + if (e) return rej(e); + res(path); + }); + }); + + // write content at URL to temp file + await new Promise((res, rej) => { + const writable = fs.createWriteStream(path); + request(url) + .on('error', rej) + .on('end', () => { + writable.close(); + res(path); + }) + .pipe(writable) + .on('error', rej); + }); + + const driveFile = await create(user, path, name, null, folderId, false, uri); + + log(`created: ${driveFile._id}`); + + // clean-up + fs.unlink(path, (e) => { + if (e) log(e.stack); + }); + + return driveFile; +}; diff --git a/src/services/following/create.ts b/src/services/following/create.ts new file mode 100644 index 0000000000..d919f4487f --- /dev/null +++ b/src/services/following/create.ts @@ -0,0 +1,72 @@ +import User, { isLocalUser, isRemoteUser, pack as packUser, IUser } from '../../models/user'; +import Following from '../../models/following'; +import FollowingLog from '../../models/following-log'; +import FollowedLog from '../../models/followed-log'; +import event from '../../publishers/stream'; +import notify from '../../publishers/notify'; +import context from '../../remote/activitypub/renderer/context'; +import renderFollow from '../../remote/activitypub/renderer/follow'; +import renderAccept from '../../remote/activitypub/renderer/accept'; +import { deliver } from '../../queue'; + +export default async function(follower: IUser, followee: IUser, activity?) { + const following = await Following.insert({ + createdAt: new Date(), + followerId: follower._id, + followeeId: followee._id + }); + + //#region Increment following count + User.update({ _id: follower._id }, { + $inc: { + followingCount: 1 + } + }); + + FollowingLog.insert({ + createdAt: following.createdAt, + userId: follower._id, + count: follower.followingCount + 1 + }); + //#endregion + + //#region Increment followers count + User.update({ _id: followee._id }, { + $inc: { + followersCount: 1 + } + }); + FollowedLog.insert({ + createdAt: following.createdAt, + userId: followee._id, + count: followee.followersCount + 1 + }); + //#endregion + + // Publish follow event + if (isLocalUser(follower)) { + packUser(followee, follower).then(packed => event(follower._id, 'follow', packed)); + } + + // Publish followed event + if (isLocalUser(followee)) { + packUser(follower, followee).then(packed => event(followee._id, 'followed', packed)), + + // 通知を作成 + notify(followee._id, follower._id, 'follow'); + } + + if (isLocalUser(follower) && isRemoteUser(followee)) { + const content = renderFollow(follower, followee); + content['@context'] = context; + + deliver(follower, content, followee.account.inbox).save(); + } + + if (isRemoteUser(follower) && isLocalUser(followee)) { + const content = renderAccept(activity); + content['@context'] = context; + + deliver(followee, content, follower.account.inbox).save(); + } +} diff --git a/src/services/following/delete.ts b/src/services/following/delete.ts new file mode 100644 index 0000000000..364a4803b9 --- /dev/null +++ b/src/services/following/delete.ts @@ -0,0 +1,64 @@ +import User, { isLocalUser, isRemoteUser, pack as packUser, IUser } from '../../models/user'; +import Following from '../../models/following'; +import FollowingLog from '../../models/following-log'; +import FollowedLog from '../../models/followed-log'; +import event from '../../publishers/stream'; +import context from '../../remote/activitypub/renderer/context'; +import renderFollow from '../../remote/activitypub/renderer/follow'; +import renderUndo from '../../remote/activitypub/renderer/undo'; +import { deliver } from '../../queue'; + +export default async function(follower: IUser, followee: IUser, activity?) { + const following = await Following.findOne({ + followerId: follower._id, + followeeId: followee._id + }); + + if (following == null) { + console.warn('フォロー解除がリクエストされましたがフォローしていませんでした'); + return; + } + + Following.remove({ + _id: following._id + }); + + //#region Decrement following count + User.update({ _id: follower._id }, { + $inc: { + followingCount: -1 + } + }); + + FollowingLog.insert({ + createdAt: following.createdAt, + userId: follower._id, + count: follower.followingCount - 1 + }); + //#endregion + + //#region Decrement followers count + User.update({ _id: followee._id }, { + $inc: { + followersCount: -1 + } + }); + FollowedLog.insert({ + createdAt: following.createdAt, + userId: followee._id, + count: followee.followersCount - 1 + }); + //#endregion + + // Publish unfollow event + if (isLocalUser(follower)) { + packUser(followee, follower).then(packed => event(follower._id, 'unfollow', packed)); + } + + if (isLocalUser(follower) && isRemoteUser(followee)) { + const content = renderUndo(renderFollow(follower, followee)); + content['@context'] = context; + + deliver(follower, content, followee.account.inbox).save(); + } +} diff --git a/src/services/post/create.ts b/src/services/post/create.ts new file mode 100644 index 0000000000..9723dbe452 --- /dev/null +++ b/src/services/post/create.ts @@ -0,0 +1,344 @@ +import Post, { pack, IPost } from '../../models/post'; +import User, { isLocalUser, IUser } from '../../models/user'; +import stream from '../../publishers/stream'; +import Following from '../../models/following'; +import { deliver } from '../../queue'; +import renderNote from '../../remote/activitypub/renderer/note'; +import renderCreate from '../../remote/activitypub/renderer/create'; +import context from '../../remote/activitypub/renderer/context'; +import { IDriveFile } from '../../models/drive-file'; +import notify from '../../publishers/notify'; +import PostWatching from '../../models/post-watching'; +import watch from './watch'; +import Mute from '../../models/mute'; +import pushSw from '../../publishers/push-sw'; +import event from '../../publishers/stream'; +import parse from '../../text/parse'; +import html from '../../text/html'; +import { IApp } from '../../models/app'; + +export default async (user: IUser, content: { + createdAt?: Date; + text?: string; + reply?: IPost; + repost?: IPost; + media?: IDriveFile[]; + geo?: any; + poll?: any; + viaMobile?: boolean; + tags?: string[]; + cw?: string; + visibility?: string; + uri?: string; + app?: IApp; +}, silent = false) => new Promise(async (res, rej) => { + if (content.createdAt == null) content.createdAt = new Date(); + if (content.visibility == null) content.visibility = 'public'; + + const tags = content.tags || []; + + let tokens = null; + + if (content.text) { + // Analyze + tokens = parse(content.text); + + // Extract hashtags + const hashtags = tokens + .filter(t => t.type == 'hashtag') + .map(t => t.hashtag); + + hashtags.forEach(tag => { + if (tags.indexOf(tag) == -1) { + tags.push(tag); + } + }); + } + + const data: any = { + createdAt: content.createdAt, + mediaIds: content.media ? content.media.map(file => file._id) : [], + replyId: content.reply ? content.reply._id : null, + repostId: content.repost ? content.repost._id : null, + text: content.text, + textHtml: tokens === null ? null : html(tokens), + poll: content.poll, + cw: content.cw, + tags, + userId: user._id, + viaMobile: content.viaMobile, + geo: content.geo || null, + appId: content.app ? content.app._id : null, + visibility: content.visibility, + + // 以下非正規化データ + _reply: content.reply ? { userId: content.reply.userId } : null, + _repost: content.repost ? { userId: content.repost.userId } : null, + }; + + if (content.uri != null) data.uri = content.uri; + + // 投稿を作成 + const post = await Post.insert(data); + + res(post); + + User.update({ _id: user._id }, { + // Increment posts count + $inc: { + postsCount: 1 + }, + // Update latest post + $set: { + latestPost: post + } + }); + + // Serialize + const postObj = await pack(post); + + // タイムラインへの投稿 + if (!post.channelId) { + // Publish event to myself's stream + if (isLocalUser(user)) { + stream(post.userId, 'post', postObj); + } + + // Fetch all followers + const followers = await Following.aggregate([{ + $lookup: { + from: 'users', + localField: 'followerId', + foreignField: '_id', + as: 'follower' + } + }, { + $match: { + followeeId: post.userId + } + }], { + _id: false + }); + + if (!silent) { + const note = await renderNote(user, post); + const content = renderCreate(note); + content['@context'] = context; + + Promise.all(followers.map(({ follower }) => { + if (isLocalUser(follower)) { + // Publish event to followers stream + stream(follower._id, 'post', postObj); + } else { + // フォロワーがリモートユーザーかつ投稿者がローカルユーザーなら投稿を配信 + if (isLocalUser(user)) { + deliver(user, content, follower.account.inbox).save(); + } + } + })); + } + } + + // チャンネルへの投稿 + /* TODO + if (post.channelId) { + promises.push( + // Increment channel index(posts count) + Channel.update({ _id: post.channelId }, { + $inc: { + index: 1 + } + }), + + // Publish event to channel + promisedPostObj.then(postObj => { + publishChannelStream(post.channelId, 'post', postObj); + }), + + Promise.all([ + promisedPostObj, + + // Get channel watchers + ChannelWatching.find({ + channelId: post.channelId, + // 削除されたドキュメントは除く + deletedAt: { $exists: false } + }) + ]).then(([postObj, watches]) => { + // チャンネルの視聴者(のタイムライン)に配信 + watches.forEach(w => { + stream(w.userId, 'post', postObj); + }); + }) + ); + }*/ + + const mentions = []; + + async function addMention(mentionee, reason) { + // Reject if already added + if (mentions.some(x => x.equals(mentionee))) return; + + // Add mention + mentions.push(mentionee); + + // Publish event + if (!user._id.equals(mentionee)) { + const mentioneeMutes = await Mute.find({ + muter_id: mentionee, + deleted_at: { $exists: false } + }); + const mentioneesMutedUserIds = mentioneeMutes.map(m => m.muteeId.toString()); + if (mentioneesMutedUserIds.indexOf(user._id.toString()) == -1) { + event(mentionee, reason, postObj); + pushSw(mentionee, reason, postObj); + } + } + } + + // If has in reply to post + if (content.reply) { + // Increment replies count + Post.update({ _id: content.reply._id }, { + $inc: { + repliesCount: 1 + } + }); + + // (自分自身へのリプライでない限りは)通知を作成 + notify(content.reply.userId, user._id, 'reply', { + postId: post._id + }); + + // Fetch watchers + PostWatching.find({ + postId: content.reply._id, + userId: { $ne: user._id }, + // 削除されたドキュメントは除く + deletedAt: { $exists: false } + }, { + fields: { + userId: true + } + }).then(watchers => { + watchers.forEach(watcher => { + notify(watcher.userId, user._id, 'reply', { + postId: post._id + }); + }); + }); + + // この投稿をWatchする + if (isLocalUser(user) && user.account.settings.autoWatch !== false) { + watch(user._id, content.reply); + } + + // Add mention + addMention(content.reply.userId, 'reply'); + } + + // If it is repost + if (content.repost) { + // Notify + const type = content.text ? 'quote' : 'repost'; + notify(content.repost.userId, user._id, type, { + post_id: post._id + }); + + // Fetch watchers + PostWatching.find({ + postId: content.repost._id, + userId: { $ne: user._id }, + // 削除されたドキュメントは除く + deletedAt: { $exists: false } + }, { + fields: { + userId: true + } + }).then(watchers => { + watchers.forEach(watcher => { + notify(watcher.userId, user._id, type, { + postId: post._id + }); + }); + }); + + // この投稿をWatchする + if (isLocalUser(user) && user.account.settings.autoWatch !== false) { + watch(user._id, content.repost); + } + + // If it is quote repost + if (content.text) { + // Add mention + addMention(content.repost.userId, 'quote'); + } else { + // Publish event + if (!user._id.equals(content.repost.userId)) { + event(content.repost.userId, 'repost', postObj); + } + } + + // 今までで同じ投稿をRepostしているか + const existRepost = await Post.findOne({ + userId: user._id, + repostId: content.repost._id, + _id: { + $ne: post._id + } + }); + + if (!existRepost) { + // Update repostee status + Post.update({ _id: content.repost._id }, { + $inc: { + repostCount: 1 + } + }); + } + } + + // If has text content + if (content.text) { + // Extract an '@' mentions + const atMentions = tokens + .filter(t => t.type == 'mention') + .map(m => m.username) + // Drop dupulicates + .filter((v, i, s) => s.indexOf(v) == i); + + // Resolve all mentions + await Promise.all(atMentions.map(async mention => { + // Fetch mentioned user + // SELECT _id + const mentionee = await User + .findOne({ + usernameLower: mention.toLowerCase() + }, { _id: true }); + + // When mentioned user not found + if (mentionee == null) return; + + // 既に言及されたユーザーに対する返信や引用repostの場合も無視 + if (content.reply && content.reply.userId.equals(mentionee._id)) return; + if (content.repost && content.repost.userId.equals(mentionee._id)) return; + + // Add mention + addMention(mentionee._id, 'mention'); + + // Create notification + notify(mentionee._id, user._id, 'mention', { + post_id: post._id + }); + })); + } + + // Append mentions data + if (mentions.length > 0) { + Post.update({ _id: post._id }, { + $set: { + mentions + } + }); + } +}); diff --git a/src/services/post/reaction/create.ts b/src/services/post/reaction/create.ts new file mode 100644 index 0000000000..e69de29bb2 diff --git a/src/services/post/watch.ts b/src/services/post/watch.ts new file mode 100644 index 0000000000..bbd9976f40 --- /dev/null +++ b/src/services/post/watch.ts @@ -0,0 +1,26 @@ +import * as mongodb from 'mongodb'; +import Watching from '../../models/post-watching'; + +export default async (me: mongodb.ObjectID, post: object) => { + // 自分の投稿はwatchできない + if (me.equals((post as any).userId)) { + return; + } + + // if watching now + const exist = await Watching.findOne({ + postId: (post as any)._id, + userId: me, + deletedAt: { $exists: false } + }); + + if (exist !== null) { + return; + } + + await Watching.insert({ + createdAt: new Date(), + postId: (post as any)._id, + userId: me + }); +}; -- cgit v1.2.3-freya From 0154e44e1d02829e8f35fa131005448f694e745e Mon Sep 17 00:00:00 2001 From: syuilo Date: Fri, 6 Apr 2018 03:42:55 +0900 Subject: Fix bugs --- src/queue/processors/http/index.ts | 3 ++- src/services/post/create.ts | 8 +++++--- 2 files changed, 7 insertions(+), 4 deletions(-) (limited to 'src/services') diff --git a/src/queue/processors/http/index.ts b/src/queue/processors/http/index.ts index 3d7d941b1a..61d7f9ac94 100644 --- a/src/queue/processors/http/index.ts +++ b/src/queue/processors/http/index.ts @@ -12,8 +12,9 @@ export default (job, done) => { const handler = handlers[job.data.type]; if (handler) { - handler(job).then(() => done(), done); + handler(job, done); } else { console.warn(`Unknown job: ${job.data.type}`); + done(); } }; diff --git a/src/services/post/create.ts b/src/services/post/create.ts index 9723dbe452..405e4a2f7b 100644 --- a/src/services/post/create.ts +++ b/src/services/post/create.ts @@ -98,7 +98,7 @@ export default async (user: IUser, content: { const postObj = await pack(post); // タイムラインへの投稿 - if (!post.channelId) { + if (post.channelId == null) { // Publish event to myself's stream if (isLocalUser(user)) { stream(post.userId, 'post', postObj); @@ -110,7 +110,7 @@ export default async (user: IUser, content: { from: 'users', localField: 'followerId', foreignField: '_id', - as: 'follower' + as: 'user' } }, { $match: { @@ -125,7 +125,9 @@ export default async (user: IUser, content: { const content = renderCreate(note); content['@context'] = context; - Promise.all(followers.map(({ follower }) => { + Promise.all(followers.map(follower => { + follower = follower.user[0]; + if (isLocalUser(follower)) { // Publish event to followers stream stream(follower._id, 'post', postObj); -- cgit v1.2.3-freya From a6fb4f2e33b5bcbda5c9461d3c55e3d1c956b2c9 Mon Sep 17 00:00:00 2001 From: syuilo Date: Fri, 6 Apr 2018 04:04:50 +0900 Subject: wip --- src/models/post.ts | 14 +++++++ src/services/post/create.ts | 98 +++++++++++++++++++++++++-------------------- 2 files changed, 69 insertions(+), 43 deletions(-) (limited to 'src/services') diff --git a/src/models/post.ts b/src/models/post.ts index 2f2b51b946..68a638fa2f 100644 --- a/src/models/post.ts +++ b/src/models/post.ts @@ -52,6 +52,20 @@ export type IPost = { speed: number; }; uri: string; + + _reply?: { + userId: mongo.ObjectID; + }; + _repost?: { + userId: mongo.ObjectID; + }; + _user: { + host: string; + hostLower: string; + account: { + inbox?: string; + }; + }; }; /** diff --git a/src/services/post/create.ts b/src/services/post/create.ts index 405e4a2f7b..745683b518 100644 --- a/src/services/post/create.ts +++ b/src/services/post/create.ts @@ -1,5 +1,5 @@ import Post, { pack, IPost } from '../../models/post'; -import User, { isLocalUser, IUser } from '../../models/user'; +import User, { isLocalUser, IUser, isRemoteUser } from '../../models/user'; import stream from '../../publishers/stream'; import Following from '../../models/following'; import { deliver } from '../../queue'; @@ -17,7 +17,7 @@ import parse from '../../text/parse'; import html from '../../text/html'; import { IApp } from '../../models/app'; -export default async (user: IUser, content: { +export default async (user: IUser, data: { createdAt?: Date; text?: string; reply?: IPost; @@ -32,16 +32,16 @@ export default async (user: IUser, content: { uri?: string; app?: IApp; }, silent = false) => new Promise(async (res, rej) => { - if (content.createdAt == null) content.createdAt = new Date(); - if (content.visibility == null) content.visibility = 'public'; + if (data.createdAt == null) data.createdAt = new Date(); + if (data.visibility == null) data.visibility = 'public'; - const tags = content.tags || []; + const tags = data.tags || []; let tokens = null; - if (content.text) { + if (data.text) { // Analyze - tokens = parse(content.text); + tokens = parse(data.text); // Extract hashtags const hashtags = tokens @@ -55,31 +55,38 @@ export default async (user: IUser, content: { }); } - const data: any = { - createdAt: content.createdAt, - mediaIds: content.media ? content.media.map(file => file._id) : [], - replyId: content.reply ? content.reply._id : null, - repostId: content.repost ? content.repost._id : null, - text: content.text, + const insert: any = { + createdAt: data.createdAt, + mediaIds: data.media ? data.media.map(file => file._id) : [], + replyId: data.reply ? data.reply._id : null, + repostId: data.repost ? data.repost._id : null, + text: data.text, textHtml: tokens === null ? null : html(tokens), - poll: content.poll, - cw: content.cw, + poll: data.poll, + cw: data.cw, tags, userId: user._id, - viaMobile: content.viaMobile, - geo: content.geo || null, - appId: content.app ? content.app._id : null, - visibility: content.visibility, + viaMobile: data.viaMobile, + geo: data.geo || null, + appId: data.app ? data.app._id : null, + visibility: data.visibility, // 以下非正規化データ - _reply: content.reply ? { userId: content.reply.userId } : null, - _repost: content.repost ? { userId: content.repost.userId } : null, + _reply: data.reply ? { userId: data.reply.userId } : null, + _repost: data.repost ? { userId: data.repost.userId } : null, + _user: { + host: user.host, + hostLower: user.hostLower, + account: isLocalUser(user) ? {} : { + inbox: user.account.inbox + } + } }; - if (content.uri != null) data.uri = content.uri; + if (data.uri != null) insert.uri = data.uri; // 投稿を作成 - const post = await Post.insert(data); + const post = await Post.insert(insert); res(post); @@ -125,6 +132,11 @@ export default async (user: IUser, content: { const content = renderCreate(note); content['@context'] = context; + // 投稿がリプライかつ投稿者がローカルユーザーかつリプライ先の投稿の投稿者がリモートユーザーなら配送 + if (data.reply && isLocalUser(user) && isRemoteUser(data.reply._user)) { + deliver(user, content, data.reply._user.account.inbox).save(); + } + Promise.all(followers.map(follower => { follower = follower.user[0]; @@ -199,22 +211,22 @@ export default async (user: IUser, content: { } // If has in reply to post - if (content.reply) { + if (data.reply) { // Increment replies count - Post.update({ _id: content.reply._id }, { + Post.update({ _id: data.reply._id }, { $inc: { repliesCount: 1 } }); // (自分自身へのリプライでない限りは)通知を作成 - notify(content.reply.userId, user._id, 'reply', { + notify(data.reply.userId, user._id, 'reply', { postId: post._id }); // Fetch watchers PostWatching.find({ - postId: content.reply._id, + postId: data.reply._id, userId: { $ne: user._id }, // 削除されたドキュメントは除く deletedAt: { $exists: false } @@ -232,24 +244,24 @@ export default async (user: IUser, content: { // この投稿をWatchする if (isLocalUser(user) && user.account.settings.autoWatch !== false) { - watch(user._id, content.reply); + watch(user._id, data.reply); } // Add mention - addMention(content.reply.userId, 'reply'); + addMention(data.reply.userId, 'reply'); } // If it is repost - if (content.repost) { + if (data.repost) { // Notify - const type = content.text ? 'quote' : 'repost'; - notify(content.repost.userId, user._id, type, { + const type = data.text ? 'quote' : 'repost'; + notify(data.repost.userId, user._id, type, { post_id: post._id }); // Fetch watchers PostWatching.find({ - postId: content.repost._id, + postId: data.repost._id, userId: { $ne: user._id }, // 削除されたドキュメントは除く deletedAt: { $exists: false } @@ -267,24 +279,24 @@ export default async (user: IUser, content: { // この投稿をWatchする if (isLocalUser(user) && user.account.settings.autoWatch !== false) { - watch(user._id, content.repost); + watch(user._id, data.repost); } // If it is quote repost - if (content.text) { + if (data.text) { // Add mention - addMention(content.repost.userId, 'quote'); + addMention(data.repost.userId, 'quote'); } else { // Publish event - if (!user._id.equals(content.repost.userId)) { - event(content.repost.userId, 'repost', postObj); + if (!user._id.equals(data.repost.userId)) { + event(data.repost.userId, 'repost', postObj); } } // 今までで同じ投稿をRepostしているか const existRepost = await Post.findOne({ userId: user._id, - repostId: content.repost._id, + repostId: data.repost._id, _id: { $ne: post._id } @@ -292,7 +304,7 @@ export default async (user: IUser, content: { if (!existRepost) { // Update repostee status - Post.update({ _id: content.repost._id }, { + Post.update({ _id: data.repost._id }, { $inc: { repostCount: 1 } @@ -301,7 +313,7 @@ export default async (user: IUser, content: { } // If has text content - if (content.text) { + if (data.text) { // Extract an '@' mentions const atMentions = tokens .filter(t => t.type == 'mention') @@ -322,8 +334,8 @@ export default async (user: IUser, content: { if (mentionee == null) return; // 既に言及されたユーザーに対する返信や引用repostの場合も無視 - if (content.reply && content.reply.userId.equals(mentionee._id)) return; - if (content.repost && content.repost.userId.equals(mentionee._id)) return; + if (data.reply && data.reply.userId.equals(mentionee._id)) return; + if (data.repost && data.repost.userId.equals(mentionee._id)) return; // Add mention addMention(mentionee._id, 'mention'); -- cgit v1.2.3-freya From 9c15c94f801de7019f014446aa3b8ea42980a1da Mon Sep 17 00:00:00 2001 From: syuilo Date: Fri, 6 Apr 2018 19:18:38 +0900 Subject: Remove silent flag --- src/remote/activitypub/act/create.ts | 6 +++--- src/services/post/create.ts | 7 +++++-- 2 files changed, 8 insertions(+), 5 deletions(-) (limited to 'src/services') diff --git a/src/remote/activitypub/act/create.ts b/src/remote/activitypub/act/create.ts index 10083995d1..1b9bad8ff5 100644 --- a/src/remote/activitypub/act/create.ts +++ b/src/remote/activitypub/act/create.ts @@ -58,7 +58,7 @@ async function createImage(resolver: Resolver, actor: IRemoteUser, image) { return await uploadFromUrl(image.url, actor); } -async function createNote(resolver: Resolver, actor: IRemoteUser, note, silent = false) { +async function createNote(resolver: Resolver, actor: IRemoteUser, note) { if ( ('attributedTo' in note && actor.account.uri !== note.attributedTo) || typeof note.id !== 'string' @@ -86,7 +86,7 @@ async function createNote(resolver: Resolver, actor: IRemoteUser, note, silent = const inReplyTo = await resolver.resolve(note.inReplyTo) as any; const actor = await resolvePerson(inReplyTo.attributedTo); if (isRemoteUser(actor)) { - reply = await createNote(resolver, actor, inReplyTo, true); + reply = await createNote(resolver, actor, inReplyTo); } } } @@ -102,5 +102,5 @@ async function createNote(resolver: Resolver, actor: IRemoteUser, note, silent = viaMobile: false, geo: undefined, uri: note.id - }, silent); + }); } diff --git a/src/services/post/create.ts b/src/services/post/create.ts index 745683b518..0bede2772d 100644 --- a/src/services/post/create.ts +++ b/src/services/post/create.ts @@ -31,7 +31,7 @@ export default async (user: IUser, data: { visibility?: string; uri?: string; app?: IApp; -}, silent = false) => new Promise(async (res, rej) => { +}) => new Promise(async (res, rej) => { if (data.createdAt == null) data.createdAt = new Date(); if (data.visibility == null) data.visibility = 'public'; @@ -127,7 +127,10 @@ export default async (user: IUser, data: { _id: false }); - if (!silent) { + // この投稿が3分以内に作成されたものであるならストリームに配信 + const shouldDistribute = new Date().getTime() - post.createdAt.getTime() < 1000 * 60 * 3; + + if (shouldDistribute) { const note = await renderNote(user, post); const content = renderCreate(note); content['@context'] = context; -- cgit v1.2.3-freya From a01251477ee5ab4766810453dd540170e65c02b2 Mon Sep 17 00:00:00 2001 From: syuilo Date: Fri, 6 Apr 2018 20:45:33 +0900 Subject: Revert "Remove silent flag" This reverts commit 9c15c94f801de7019f014446aa3b8ea42980a1da. --- src/remote/activitypub/act/create/note.ts | 2 +- src/services/post/create.ts | 7 ++----- 2 files changed, 3 insertions(+), 6 deletions(-) (limited to 'src/services') diff --git a/src/remote/activitypub/act/create/note.ts b/src/remote/activitypub/act/create/note.ts index 2ccd503aeb..d50042e163 100644 --- a/src/remote/activitypub/act/create/note.ts +++ b/src/remote/activitypub/act/create/note.ts @@ -10,7 +10,7 @@ import createImage from './image'; const log = debug('misskey:activitypub'); -export default async function createNote(resolver: Resolver, actor: IRemoteUser, note): Promise { +export default async function createNote(resolver: Resolver, actor: IRemoteUser, note, silent = false): Promise { if ( ('attributedTo' in note && actor.account.uri !== note.attributedTo) || typeof note.id !== 'string' diff --git a/src/services/post/create.ts b/src/services/post/create.ts index 0bede2772d..745683b518 100644 --- a/src/services/post/create.ts +++ b/src/services/post/create.ts @@ -31,7 +31,7 @@ export default async (user: IUser, data: { visibility?: string; uri?: string; app?: IApp; -}) => new Promise(async (res, rej) => { +}, silent = false) => new Promise(async (res, rej) => { if (data.createdAt == null) data.createdAt = new Date(); if (data.visibility == null) data.visibility = 'public'; @@ -127,10 +127,7 @@ export default async (user: IUser, data: { _id: false }); - // この投稿が3分以内に作成されたものであるならストリームに配信 - const shouldDistribute = new Date().getTime() - post.createdAt.getTime() < 1000 * 60 * 3; - - if (shouldDistribute) { + if (!silent) { const note = await renderNote(user, post); const content = renderCreate(note); content['@context'] = context; -- cgit v1.2.3-freya