diff options
Diffstat (limited to 'src/api')
| -rw-r--r-- | src/api/drive/add-file.ts | 314 | ||||
| -rw-r--r-- | src/api/drive/upload-from-url.ts | 46 | ||||
| -rw-r--r-- | src/api/following/create.ts | 82 | ||||
| -rw-r--r-- | src/api/post/create.ts | 149 | ||||
| -rw-r--r-- | src/api/post/distribute.ts | 190 | ||||
| -rw-r--r-- | src/api/post/watch.ts | 26 |
6 files changed, 807 insertions, 0 deletions
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<string> => 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<any> => + 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<string> => 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<number> => 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<any>((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<any> => 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<IDriveFile> => { + 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<Promise<any>>; + + 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 + }); +}; |