From 77f056b4fcdf74da8b6a8cc4a923eb8789d6f5ae Mon Sep 17 00:00:00 2001 From: syuilo Date: Wed, 4 Apr 2018 23:59:38 +0900 Subject: wip --- src/api/drive/add-file.ts | 314 ++++++++++++++++++++++++++++++++++ src/api/drive/upload-from-url.ts | 46 +++++ src/api/following/create.ts | 82 +++++++++ src/api/post/create.ts | 149 ++++++++++++++++ src/api/post/distribute.ts | 190 ++++++++++++++++++++ src/api/post/watch.ts | 26 +++ src/drive/add-file.ts | 314 ---------------------------------- src/drive/upload-from-url.ts | 46 ----- src/post/create.ts | 149 ---------------- src/post/distribute.ts | 190 -------------------- src/post/watch.ts | 26 --- src/queue/processors/http/unfollow.ts | 97 ++++++----- src/remote/activitypub/act/create.ts | 8 +- src/remote/activitypub/act/follow.ts | 59 +------ 14 files changed, 867 insertions(+), 829 deletions(-) create mode 100644 src/api/drive/add-file.ts create mode 100644 src/api/drive/upload-from-url.ts create mode 100644 src/api/following/create.ts create mode 100644 src/api/post/create.ts create mode 100644 src/api/post/distribute.ts create mode 100644 src/api/post/watch.ts delete mode 100644 src/drive/add-file.ts delete mode 100644 src/drive/upload-from-url.ts delete mode 100644 src/post/create.ts delete mode 100644 src/post/distribute.ts delete mode 100644 src/post/watch.ts (limited to 'src') diff --git a/src/api/drive/add-file.ts b/src/api/drive/add-file.ts new file mode 100644 index 0000000000..24eb5208d5 --- /dev/null +++ b/src/api/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/api/drive/upload-from-url.ts b/src/api/drive/upload-from-url.ts new file mode 100644 index 0000000000..f96af0f266 --- /dev/null +++ b/src/api/drive/upload-from-url.ts @@ -0,0 +1,46 @@ +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:common:drive:upload_from_url'); + +export default async (url, user, folderId = null, uri = null): Promise => { + let name = URL.parse(url).pathname.split('/').pop(); + if (!validateFileName(name)) { + name = null; + } + + // 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); + + // clean-up + fs.unlink(path, (e) => { + if (e) log(e.stack); + }); + + return driveFile; +}; diff --git a/src/api/following/create.ts b/src/api/following/create.ts new file mode 100644 index 0000000000..353a6c8920 --- /dev/null +++ b/src/api/following/create.ts @@ -0,0 +1,82 @@ +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 { createHttp } 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; + + createHttp({ + type: 'deliver', + user: follower, + content, + to: followee.account.inbox + }).save(); + } + + if (isRemoteUser(follower) && isLocalUser(followee)) { + const content = renderAccept(activity); + content['@context'] = context; + + createHttp({ + type: 'deliver', + user: followee, + content, + to: follower.account.inbox + }).save(); + } +} diff --git a/src/api/post/create.ts b/src/api/post/create.ts new file mode 100644 index 0000000000..f78bbe7521 --- /dev/null +++ b/src/api/post/create.ts @@ -0,0 +1,149 @@ +import parseAcct from '../acct/parse'; +import Post, { pack } from '../models/post'; +import User, { isLocalUser, isRemoteUser, IUser } from '../models/user'; +import stream from '../publishers/stream'; +import Following from '../models/following'; +import { createHttp } from '../queue'; +import renderNote from '../remote/activitypub/renderer/note'; +import renderCreate from '../remote/activitypub/renderer/create'; +import context from '../remote/activitypub/renderer/context'; + +export default async (user: IUser, post, reply, repost, atMentions) => { + post.mentions = []; + + function addMention(mentionee) { + // Reject if already added + if (post.mentions.some(x => x.equals(mentionee))) return; + + // Add mention + post.mentions.push(mentionee); + } + + if (reply) { + // Add mention + addMention(reply.userId); + post.replyId = reply._id; + post._reply = { userId: reply.userId }; + } else { + post.replyId = null; + post._reply = null; + } + + if (repost) { + if (post.text) { + // Add mention + addMention(repost.userId); + } + + post.repostId = repost._id; + post._repost = { userId: repost.userId }; + } else { + post.repostId = null; + post._repost = null; + } + + await Promise.all(atMentions.map(async mention => { + // Fetch mentioned user + // SELECT _id + const { _id } = await User + .findOne(parseAcct(mention), { _id: true }); + + // Add mention + addMention(_id); + })); + + const inserted = await Post.insert(post); + + User.update({ _id: user._id }, { + // Increment my posts count + $inc: { + postsCount: 1 + }, + + $set: { + latestPost: post._id + } + }); + + const postObj = await pack(inserted); + + // タイムラインへの投稿 + if (!post.channelId) { + // Publish event to myself's stream + 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 + }); + + 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)) { + createHttp({ + type: 'deliver', + user, + content, + to: 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); + }); + }) + ); + }*/ + + return Promise.all(promises); + +}; diff --git a/src/api/post/distribute.ts b/src/api/post/distribute.ts new file mode 100644 index 0000000000..49c6eb22df --- /dev/null +++ b/src/api/post/distribute.ts @@ -0,0 +1,190 @@ +import Mute from '../models/mute'; +import Post, { pack } from '../models/post'; +import Watching from '../models/post-watching'; +import User from '../models/user'; +import stream from '../publishers/stream'; +import notify from '../publishers/notify'; +import pushSw from '../publishers/push-sw'; +import queue from '../queue'; +import watch from './watch'; + +export default async (user, mentions, post) => { + const promisedPostObj = pack(post); + const promises = [ + User.update({ _id: user._id }, { + // Increment my posts count + $inc: { + postsCount: 1 + }, + + $set: { + latestPost: post._id + } + }), + new Promise((resolve, reject) => queue.create('http', { + type: 'deliverPost', + id: post._id, + }).save(error => error ? reject(error) : resolve())), + ] as Array>; + + function addMention(promisedMentionee, reason) { + // Publish event + promises.push(promisedMentionee.then(mentionee => { + if (user._id.equals(mentionee)) { + return Promise.resolve(); + } + + return Promise.all([ + promisedPostObj, + Mute.find({ + muterId: mentionee, + deletedAt: { $exists: false } + }) + ]).then(([postObj, mentioneeMutes]) => { + const mentioneesMutedUserIds = mentioneeMutes.map(m => m.muteeId.toString()); + if (mentioneesMutedUserIds.indexOf(user._id.toString()) == -1) { + stream(mentionee, reason, postObj); + pushSw(mentionee, reason, postObj); + } + }); + })); + } + + // If has in reply to post + if (post.replyId) { + promises.push( + // Increment replies count + Post.update({ _id: post.replyId }, { + $inc: { + repliesCount: 1 + } + }), + + // 自分自身へのリプライでない限りは通知を作成 + promisedPostObj.then(({ reply }) => { + return notify(reply.userId, user._id, 'reply', { + postId: post._id + }); + }), + + // Fetch watchers + Watching + .find({ + postId: post.replyId, + userId: { $ne: user._id }, + // 削除されたドキュメントは除く + deletedAt: { $exists: false } + }, { + fields: { + userId: true + } + }) + .then(watchers => { + watchers.forEach(watcher => { + notify(watcher.userId, user._id, 'reply', { + postId: post._id + }); + }); + }) + ); + + // Add mention + addMention(promisedPostObj.then(({ reply }) => reply.userId), 'reply'); + + // この投稿をWatchする + if (user.account.settings.autoWatch !== false) { + promises.push(promisedPostObj.then(({ reply }) => { + return watch(user._id, reply); + })); + } + } + + // If it is repost + if (post.repostId) { + const type = post.text ? 'quote' : 'repost'; + + promises.push( + promisedPostObj.then(({ repost }) => Promise.all([ + // Notify + notify(repost.userId, user._id, type, { + postId: post._id + }), + + // この投稿をWatchする + // TODO: ユーザーが「Repostしたときに自動でWatchする」設定を + // オフにしていた場合はしない + watch(user._id, repost) + ])), + + // Fetch watchers + Watching + .find({ + postId: post.repostId, + userId: { $ne: user._id }, + // 削除されたドキュメントは除く + deletedAt: { $exists: false } + }, { + fields: { + userId: true + } + }) + .then(watchers => { + watchers.forEach(watcher => { + notify(watcher.userId, user._id, type, { + postId: post._id + }); + }); + }) + ); + + // If it is quote repost + if (post.text) { + // Add mention + addMention(promisedPostObj.then(({ repost }) => repost.userId), 'quote'); + } else { + promises.push(promisedPostObj.then(postObj => { + // Publish event + if (!user._id.equals(postObj.repost.userId)) { + stream(postObj.repost.userId, 'repost', postObj); + } + })); + } + + // 今までで同じ投稿をRepostしているか + const existRepost = await Post.findOne({ + userId: user._id, + repostId: post.repostId, + _id: { + $ne: post._id + } + }); + + if (!existRepost) { + // Update repostee status + promises.push(Post.update({ _id: post.repostId }, { + $inc: { + repostCount: 1 + } + })); + } + } + + // Resolve all mentions + await promisedPostObj.then(({ reply, repost }) => Promise.all(mentions.map(async mention => { + // 既に言及されたユーザーに対する返信や引用repostの場合も無視 + if (reply && reply.userId.equals(mention)) return; + if (repost && repost.userId.equals(mention)) return; + + // Add mention + addMention(mention, 'mention'); + + // Create notification + await notify(mention, user._id, 'mention', { + postId: post._id + }); + }))); + + await Promise.all(promises); + + return promisedPostObj; +}; diff --git a/src/api/post/watch.ts b/src/api/post/watch.ts new file mode 100644 index 0000000000..61ea444430 --- /dev/null +++ b/src/api/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 + }); +}; diff --git a/src/drive/add-file.ts b/src/drive/add-file.ts deleted file mode 100644 index 24eb5208d5..0000000000 --- a/src/drive/add-file.ts +++ /dev/null @@ -1,314 +0,0 @@ -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/drive/upload-from-url.ts b/src/drive/upload-from-url.ts deleted file mode 100644 index f96af0f266..0000000000 --- a/src/drive/upload-from-url.ts +++ /dev/null @@ -1,46 +0,0 @@ -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:common:drive:upload_from_url'); - -export default async (url, user, folderId = null, uri = null): Promise => { - let name = URL.parse(url).pathname.split('/').pop(); - if (!validateFileName(name)) { - name = null; - } - - // 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); - - // clean-up - fs.unlink(path, (e) => { - if (e) log(e.stack); - }); - - return driveFile; -}; diff --git a/src/post/create.ts b/src/post/create.ts deleted file mode 100644 index f78bbe7521..0000000000 --- a/src/post/create.ts +++ /dev/null @@ -1,149 +0,0 @@ -import parseAcct from '../acct/parse'; -import Post, { pack } from '../models/post'; -import User, { isLocalUser, isRemoteUser, IUser } from '../models/user'; -import stream from '../publishers/stream'; -import Following from '../models/following'; -import { createHttp } from '../queue'; -import renderNote from '../remote/activitypub/renderer/note'; -import renderCreate from '../remote/activitypub/renderer/create'; -import context from '../remote/activitypub/renderer/context'; - -export default async (user: IUser, post, reply, repost, atMentions) => { - post.mentions = []; - - function addMention(mentionee) { - // Reject if already added - if (post.mentions.some(x => x.equals(mentionee))) return; - - // Add mention - post.mentions.push(mentionee); - } - - if (reply) { - // Add mention - addMention(reply.userId); - post.replyId = reply._id; - post._reply = { userId: reply.userId }; - } else { - post.replyId = null; - post._reply = null; - } - - if (repost) { - if (post.text) { - // Add mention - addMention(repost.userId); - } - - post.repostId = repost._id; - post._repost = { userId: repost.userId }; - } else { - post.repostId = null; - post._repost = null; - } - - await Promise.all(atMentions.map(async mention => { - // Fetch mentioned user - // SELECT _id - const { _id } = await User - .findOne(parseAcct(mention), { _id: true }); - - // Add mention - addMention(_id); - })); - - const inserted = await Post.insert(post); - - User.update({ _id: user._id }, { - // Increment my posts count - $inc: { - postsCount: 1 - }, - - $set: { - latestPost: post._id - } - }); - - const postObj = await pack(inserted); - - // タイムラインへの投稿 - if (!post.channelId) { - // Publish event to myself's stream - 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 - }); - - 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)) { - createHttp({ - type: 'deliver', - user, - content, - to: 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); - }); - }) - ); - }*/ - - return Promise.all(promises); - -}; diff --git a/src/post/distribute.ts b/src/post/distribute.ts deleted file mode 100644 index 49c6eb22df..0000000000 --- a/src/post/distribute.ts +++ /dev/null @@ -1,190 +0,0 @@ -import Mute from '../models/mute'; -import Post, { pack } from '../models/post'; -import Watching from '../models/post-watching'; -import User from '../models/user'; -import stream from '../publishers/stream'; -import notify from '../publishers/notify'; -import pushSw from '../publishers/push-sw'; -import queue from '../queue'; -import watch from './watch'; - -export default async (user, mentions, post) => { - const promisedPostObj = pack(post); - const promises = [ - User.update({ _id: user._id }, { - // Increment my posts count - $inc: { - postsCount: 1 - }, - - $set: { - latestPost: post._id - } - }), - new Promise((resolve, reject) => queue.create('http', { - type: 'deliverPost', - id: post._id, - }).save(error => error ? reject(error) : resolve())), - ] as Array>; - - function addMention(promisedMentionee, reason) { - // Publish event - promises.push(promisedMentionee.then(mentionee => { - if (user._id.equals(mentionee)) { - return Promise.resolve(); - } - - return Promise.all([ - promisedPostObj, - Mute.find({ - muterId: mentionee, - deletedAt: { $exists: false } - }) - ]).then(([postObj, mentioneeMutes]) => { - const mentioneesMutedUserIds = mentioneeMutes.map(m => m.muteeId.toString()); - if (mentioneesMutedUserIds.indexOf(user._id.toString()) == -1) { - stream(mentionee, reason, postObj); - pushSw(mentionee, reason, postObj); - } - }); - })); - } - - // If has in reply to post - if (post.replyId) { - promises.push( - // Increment replies count - Post.update({ _id: post.replyId }, { - $inc: { - repliesCount: 1 - } - }), - - // 自分自身へのリプライでない限りは通知を作成 - promisedPostObj.then(({ reply }) => { - return notify(reply.userId, user._id, 'reply', { - postId: post._id - }); - }), - - // Fetch watchers - Watching - .find({ - postId: post.replyId, - userId: { $ne: user._id }, - // 削除されたドキュメントは除く - deletedAt: { $exists: false } - }, { - fields: { - userId: true - } - }) - .then(watchers => { - watchers.forEach(watcher => { - notify(watcher.userId, user._id, 'reply', { - postId: post._id - }); - }); - }) - ); - - // Add mention - addMention(promisedPostObj.then(({ reply }) => reply.userId), 'reply'); - - // この投稿をWatchする - if (user.account.settings.autoWatch !== false) { - promises.push(promisedPostObj.then(({ reply }) => { - return watch(user._id, reply); - })); - } - } - - // If it is repost - if (post.repostId) { - const type = post.text ? 'quote' : 'repost'; - - promises.push( - promisedPostObj.then(({ repost }) => Promise.all([ - // Notify - notify(repost.userId, user._id, type, { - postId: post._id - }), - - // この投稿をWatchする - // TODO: ユーザーが「Repostしたときに自動でWatchする」設定を - // オフにしていた場合はしない - watch(user._id, repost) - ])), - - // Fetch watchers - Watching - .find({ - postId: post.repostId, - userId: { $ne: user._id }, - // 削除されたドキュメントは除く - deletedAt: { $exists: false } - }, { - fields: { - userId: true - } - }) - .then(watchers => { - watchers.forEach(watcher => { - notify(watcher.userId, user._id, type, { - postId: post._id - }); - }); - }) - ); - - // If it is quote repost - if (post.text) { - // Add mention - addMention(promisedPostObj.then(({ repost }) => repost.userId), 'quote'); - } else { - promises.push(promisedPostObj.then(postObj => { - // Publish event - if (!user._id.equals(postObj.repost.userId)) { - stream(postObj.repost.userId, 'repost', postObj); - } - })); - } - - // 今までで同じ投稿をRepostしているか - const existRepost = await Post.findOne({ - userId: user._id, - repostId: post.repostId, - _id: { - $ne: post._id - } - }); - - if (!existRepost) { - // Update repostee status - promises.push(Post.update({ _id: post.repostId }, { - $inc: { - repostCount: 1 - } - })); - } - } - - // Resolve all mentions - await promisedPostObj.then(({ reply, repost }) => Promise.all(mentions.map(async mention => { - // 既に言及されたユーザーに対する返信や引用repostの場合も無視 - if (reply && reply.userId.equals(mention)) return; - if (repost && repost.userId.equals(mention)) return; - - // Add mention - addMention(mention, 'mention'); - - // Create notification - await notify(mention, user._id, 'mention', { - postId: post._id - }); - }))); - - await Promise.all(promises); - - return promisedPostObj; -}; diff --git a/src/post/watch.ts b/src/post/watch.ts deleted file mode 100644 index 61ea444430..0000000000 --- a/src/post/watch.ts +++ /dev/null @@ -1,26 +0,0 @@ -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 - }); -}; diff --git a/src/queue/processors/http/unfollow.ts b/src/queue/processors/http/unfollow.ts index d3d5f2246f..801a3612a7 100644 --- a/src/queue/processors/http/unfollow.ts +++ b/src/queue/processors/http/unfollow.ts @@ -1,56 +1,63 @@ -import FollowedLog from '../../models/followed-log'; -import Following from '../../models/following'; -import FollowingLog from '../../models/following-log'; -import User, { isRemoteUser, pack as packUser } from '../../models/user'; -import stream from '../../publishers/stream'; -import renderFollow from '../../remote/activitypub/renderer/follow'; -import renderUndo from '../../remote/activitypub/renderer/undo'; -import context from '../../remote/activitypub/renderer/context'; -import request from '../../remote/request'; +import FollowedLog from '../../../models/followed-log'; +import Following from '../../../models/following'; +import FollowingLog from '../../../models/following-log'; +import User, { isLocalUser, isRemoteUser, pack as packUser } from '../../../models/user'; +import stream from '../../../publishers/stream'; +import renderFollow from '../../../remote/activitypub/renderer/follow'; +import renderUndo from '../../../remote/activitypub/renderer/undo'; +import context from '../../../remote/activitypub/renderer/context'; +import request from '../../../remote/request'; +import Logger from '../../../utils/logger'; export default async ({ data }) => { - // Delete following - const following = await Following.findOneAndDelete({ _id: data.id }); + const following = await Following.findOne({ _id: data.id }); if (following === null) { return; } - const promisedFollower = User.findOne({ _id: following.followerId }); - const promisedFollowee = User.findOne({ _id: following.followeeId }); - - await Promise.all([ - // Decrement following count - User.update({ _id: following.followerId }, { $inc: { followingCount: -1 } }), - promisedFollower.then(({ followingCount }) => FollowingLog.insert({ - createdAt: new Date(), - userId: following.followerId, - count: followingCount - 1 - })), - - // Decrement followers count - User.update({ _id: following.followeeId }, { $inc: { followersCount: -1 } }), - promisedFollowee.then(({ followersCount }) => FollowedLog.insert({ - createdAt: new Date(), - userId: following.followeeId, - count: followersCount - 1 - })), - - // Publish follow event - Promise.all([promisedFollower, promisedFollowee]).then(async ([follower, followee]) => { - if (isRemoteUser(follower)) { - return; - } + const [follower, followee] = await Promise.all([ + User.findOne({ _id: following.followerId }), + User.findOne({ _id: following.followeeId }) + ]); - const promisedPackedUser = packUser(followee, follower); + if (isLocalUser(follower) && isRemoteUser(followee)) { + const undo = renderUndo(renderFollow(follower, followee)); + undo['@context'] = context; - if (isRemoteUser(followee)) { - const undo = renderUndo(renderFollow(follower, followee)); - undo['@context'] = context; + await request(follower, followee.account.inbox, undo); + } - await request(follower, followee.account.inbox, undo); - } + try { + await Promise.all([ + // Delete following + Following.findOneAndDelete({ _id: data.id }), + + // Decrement following count + User.update({ _id: follower._id }, { $inc: { followingCount: -1 } }), + FollowingLog.insert({ + createdAt: new Date(), + userId: follower._id, + count: follower.followingCount - 1 + }), + + // Decrement followers count + User.update({ _id: followee._id }, { $inc: { followersCount: -1 } }), + FollowedLog.insert({ + createdAt: new Date(), + userId: followee._id, + count: followee.followersCount - 1 + }) + ]); + + if (isLocalUser(follower)) { + return; + } + + const promisedPackedUser = packUser(followee, follower); - stream(follower._id, 'unfollow', promisedPackedUser); - }) - ]); + // Publish follow event + stream(follower._id, 'unfollow', promisedPackedUser); + } catch (error) { + Logger.error(error.toString()); + } }; diff --git a/src/remote/activitypub/act/create.ts b/src/remote/activitypub/act/create.ts index c1a30ce7d0..7ee9f8dfb7 100644 --- a/src/remote/activitypub/act/create.ts +++ b/src/remote/activitypub/act/create.ts @@ -4,10 +4,10 @@ const createDOMPurify = require('dompurify'); import Resolver from '../resolver'; import DriveFile from '../../../models/drive-file'; import Post from '../../../models/post'; -import uploadFromUrl from '../../../drive/upload-from-url'; -import createPost from '../../../post/create'; +import uploadFromUrl from '../../../api/drive/upload-from-url'; +import createPost from '../../../api/post/create'; -export default async (resolver: Resolver, actor, activity): Promise => { +export default async (actor, activity): Promise => { if ('actor' in activity && actor.account.uri !== activity.actor) { throw new Error('invalid actor'); } @@ -31,6 +31,8 @@ export default async (resolver: Resolver, actor, activity): Promise => { throw new Error(`already registered: ${uri}`); } + const resolver = new Resolver(); + const object = await resolver.resolve(activity); switch (object.type) { diff --git a/src/remote/activitypub/act/follow.ts b/src/remote/activitypub/act/follow.ts index 23fa41df8e..dc173a0acb 100644 --- a/src/remote/activitypub/act/follow.ts +++ b/src/remote/activitypub/act/follow.ts @@ -1,15 +1,9 @@ -import { MongoError } from 'mongodb'; import parseAcct from '../../../acct/parse'; -import Following, { IFollowing } from '../../../models/following'; import User from '../../../models/user'; import config from '../../../config'; -import queue from '../../../queue'; -import context from '../renderer/context'; -import renderAccept from '../renderer/accept'; -import request from '../../request'; -import Resolver from '../resolver'; +import follow from '../../../api/following/create'; -export default async (resolver: Resolver, actor, activity, distribute) => { +export default async (actor, activity): Promise => { const prefix = config.url + '/@'; const id = activity.object.id || activity.object; @@ -27,52 +21,5 @@ export default async (resolver: Resolver, actor, activity, distribute) => { throw new Error(); } - if (!distribute) { - const { _id } = await Following.findOne({ - followerId: actor._id, - followeeId: followee._id - }); - - return { - resolver, - object: { $ref: 'following', $id: _id } - }; - } - - const promisedFollowing = Following.insert({ - createdAt: new Date(), - followerId: actor._id, - followeeId: followee._id - }).then(following => new Promise((resolve, reject) => { - queue.create('http', { - type: 'follow', - following: following._id - }).save(error => { - if (error) { - reject(error); - } else { - resolve(following); - } - }); - }) as Promise, async error => { - // duplicate key error - if (error instanceof MongoError && error.code === 11000) { - return Following.findOne({ - followerId: actor._id, - followeeId: followee._id - }); - } - - throw error; - }); - - const accept = renderAccept(activity); - accept['@context'] = context; - - await request(followee, actor.account.inbox, accept); - - return promisedFollowing.then(({ _id }) => ({ - resolver, - object: { $ref: 'following', $id: _id } - })); + await follow(actor, followee, activity); }; -- cgit v1.2.3-freya