From e8b42d7e1668679e6a6ee0a7aea1e2ff7f37005b Mon Sep 17 00:00:00 2001 From: syuilo Date: Wed, 4 Apr 2018 23:12:35 +0900 Subject: wip --- src/queue/processors/db/delete-post-dependents.ts | 22 +++++++ src/queue/processors/db/index.ts | 7 +++ src/queue/processors/http/deliver.ts | 17 ++++++ src/queue/processors/http/follow.ts | 69 ++++++++++++++++++++++ src/queue/processors/http/index.ts | 17 ++++++ src/queue/processors/http/perform-activitypub.ts | 7 +++ src/queue/processors/http/process-inbox.ts | 55 +++++++++++++++++ src/queue/processors/http/report-github-failure.ts | 24 ++++++++ src/queue/processors/http/unfollow.ts | 56 ++++++++++++++++++ src/queue/processors/index.ts | 18 ++++++ 10 files changed, 292 insertions(+) create mode 100644 src/queue/processors/db/delete-post-dependents.ts create mode 100644 src/queue/processors/db/index.ts create mode 100644 src/queue/processors/http/deliver.ts create mode 100644 src/queue/processors/http/follow.ts create mode 100644 src/queue/processors/http/index.ts create mode 100644 src/queue/processors/http/perform-activitypub.ts create mode 100644 src/queue/processors/http/process-inbox.ts create mode 100644 src/queue/processors/http/report-github-failure.ts create mode 100644 src/queue/processors/http/unfollow.ts create mode 100644 src/queue/processors/index.ts (limited to 'src/queue/processors') diff --git a/src/queue/processors/db/delete-post-dependents.ts b/src/queue/processors/db/delete-post-dependents.ts new file mode 100644 index 0000000000..879c41ec9c --- /dev/null +++ b/src/queue/processors/db/delete-post-dependents.ts @@ -0,0 +1,22 @@ +import Favorite from '../../models/favorite'; +import Notification from '../../models/notification'; +import PollVote from '../../models/poll-vote'; +import PostReaction from '../../models/post-reaction'; +import PostWatching from '../../models/post-watching'; +import Post from '../../models/post'; + +export default async ({ data }) => Promise.all([ + Favorite.remove({ postId: data._id }), + Notification.remove({ postId: data._id }), + PollVote.remove({ postId: data._id }), + PostReaction.remove({ postId: data._id }), + PostWatching.remove({ postId: data._id }), + Post.find({ repostId: data._id }).then(reposts => Promise.all([ + Notification.remove({ + postId: { + $in: reposts.map(({ _id }) => _id) + } + }), + Post.remove({ repostId: data._id }) + ])) +]); diff --git a/src/queue/processors/db/index.ts b/src/queue/processors/db/index.ts new file mode 100644 index 0000000000..75838c099b --- /dev/null +++ b/src/queue/processors/db/index.ts @@ -0,0 +1,7 @@ +import deletePostDependents from './delete-post-dependents'; + +const handlers = { + deletePostDependents +}; + +export default (job, done) => handlers[job.data.type](job).then(() => done(), done); diff --git a/src/queue/processors/http/deliver.ts b/src/queue/processors/http/deliver.ts new file mode 100644 index 0000000000..8cd9eb624e --- /dev/null +++ b/src/queue/processors/http/deliver.ts @@ -0,0 +1,17 @@ +import * as kue from 'kue'; + +import Channel from '../../models/channel'; +import Following from '../../models/following'; +import ChannelWatching from '../../models/channel-watching'; +import Post, { pack } from '../../models/post'; +import User, { isLocalUser } from '../../models/user'; +import stream, { publishChannelStream } from '../../publishers/stream'; +import context from '../../remote/activitypub/renderer/context'; +import renderCreate from '../../remote/activitypub/renderer/create'; +import renderNote from '../../remote/activitypub/renderer/note'; +import request from '../../remote/request'; + +export default async (job: kue.Job, done): Promise => { + + request(user, following.follower[0].account.inbox, create); +} diff --git a/src/queue/processors/http/follow.ts b/src/queue/processors/http/follow.ts new file mode 100644 index 0000000000..8bf890efbc --- /dev/null +++ b/src/queue/processors/http/follow.ts @@ -0,0 +1,69 @@ +import User, { isLocalUser, pack as packUser } 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 render from '../../remote/activitypub/renderer/follow'; +import request from '../../remote/request'; + +export default ({ data }) => Following.findOne({ _id: data.following }).then(({ followerId, followeeId }) => { + const promisedFollower = User.findOne({ _id: followerId }); + const promisedFollowee = User.findOne({ _id: followeeId }); + + return Promise.all([ + // Increment following count + User.update(followerId, { + $inc: { + followingCount: 1 + } + }), + + promisedFollower.then(({ followingCount }) => FollowingLog.insert({ + createdAt: data.following.createdAt, + userId: followerId, + count: followingCount + 1 + })), + + // Increment followers count + User.update({ _id: followeeId }, { + $inc: { + followersCount: 1 + } + }), + + promisedFollowee.then(({ followersCount }) => FollowedLog.insert({ + createdAt: data.following.createdAt, + userId: followerId, + count: followersCount + 1 + })), + + // Notify + promisedFollowee.then(followee => followee.host === null ? + notify(followeeId, followerId, 'follow') : null), + + // Publish follow event + Promise.all([promisedFollower, promisedFollowee]).then(([follower, followee]) => { + let followerEvent; + let followeeEvent; + + if (isLocalUser(follower)) { + followerEvent = packUser(followee, follower) + .then(packed => event(follower._id, 'follow', packed)); + } + + if (isLocalUser(followee)) { + followeeEvent = packUser(follower, followee) + .then(packed => event(followee._id, 'followed', packed)); + } else if (isLocalUser(follower)) { + const rendered = render(follower, followee); + rendered['@context'] = context; + + followeeEvent = request(follower, followee.account.inbox, rendered); + } + + return Promise.all([followerEvent, followeeEvent]); + }) + ]); +}); diff --git a/src/queue/processors/http/index.ts b/src/queue/processors/http/index.ts new file mode 100644 index 0000000000..8f9aa717c3 --- /dev/null +++ b/src/queue/processors/http/index.ts @@ -0,0 +1,17 @@ +import deliverPost from './deliver-post'; +import follow from './follow'; +import performActivityPub from './perform-activitypub'; +import processInbox from './process-inbox'; +import reportGitHubFailure from './report-github-failure'; +import unfollow from './unfollow'; + +const handlers = { + deliverPost, + follow, + performActivityPub, + processInbox, + reportGitHubFailure, + unfollow +}; + +export default (job, done) => handlers[job.data.type](job).then(() => done(), done); diff --git a/src/queue/processors/http/perform-activitypub.ts b/src/queue/processors/http/perform-activitypub.ts new file mode 100644 index 0000000000..963e532fe5 --- /dev/null +++ b/src/queue/processors/http/perform-activitypub.ts @@ -0,0 +1,7 @@ +import User from '../../models/user'; +import act from '../../remote/activitypub/act'; +import Resolver from '../../remote/activitypub/resolver'; + +export default ({ data }) => User.findOne({ _id: data.actor }) + .then(actor => act(new Resolver(), actor, data.outbox)) + .then(Promise.all); diff --git a/src/queue/processors/http/process-inbox.ts b/src/queue/processors/http/process-inbox.ts new file mode 100644 index 0000000000..fff1fbf663 --- /dev/null +++ b/src/queue/processors/http/process-inbox.ts @@ -0,0 +1,55 @@ +import * as kue from 'kue'; + +import { verifySignature } from 'http-signature'; +import parseAcct from '../../acct/parse'; +import User, { IRemoteUser } from '../../models/user'; +import act from '../../remote/activitypub/act'; +import resolvePerson from '../../remote/activitypub/resolve-person'; +import Resolver from '../../remote/activitypub/resolver'; + +// ユーザーのinboxにアクティビティが届いた時の処理 +export default async (job: kue.Job, done): Promise => { + const signature = job.data.signature; + const activity = job.data.activity; + + const keyIdLower = signature.keyId.toLowerCase(); + let user; + + if (keyIdLower.startsWith('acct:')) { + const { username, host } = parseAcct(keyIdLower.slice('acct:'.length)); + if (host === null) { + console.warn(`request was made by local user: @${username}`); + done(); + } + + user = await User.findOne({ usernameLower: username, hostLower: host }) as IRemoteUser; + } else { + user = await User.findOne({ + host: { $ne: null }, + 'account.publicKey.id': signature.keyId + }) as IRemoteUser; + + // アクティビティを送信してきたユーザーがまだMisskeyサーバーに登録されていなかったら登録する + if (user === null) { + user = await resolvePerson(signature.keyId); + } + } + + if (user === null) { + done(new Error('failed to resolve user')); + return; + } + + if (!verifySignature(signature, user.account.publicKey.publicKeyPem)) { + done(new Error('signature verification failed')); + return; + } + + // アクティビティを処理 + try { + await act(new Resolver(), user, activity); + done(); + } catch (e) { + done(e); + } +}; diff --git a/src/queue/processors/http/report-github-failure.ts b/src/queue/processors/http/report-github-failure.ts new file mode 100644 index 0000000000..4f6f5ccee5 --- /dev/null +++ b/src/queue/processors/http/report-github-failure.ts @@ -0,0 +1,24 @@ +import * as request from 'request-promise-native'; +import User from '../../models/user'; +const createPost = require('../../server/api/endpoints/posts/create'); + +export default async ({ data }) => { + const asyncBot = User.findOne({ _id: data.userId }); + + // Fetch parent status + const parentStatuses = await request({ + url: `${data.parentUrl}/statuses`, + headers: { + 'User-Agent': 'misskey' + }, + json: true + }); + + const parentState = parentStatuses[0].state; + const stillFailed = parentState == 'failure' || parentState == 'error'; + const text = stillFailed ? + `**⚠️BUILD STILL FAILED⚠️**: ?[${data.message}](${data.htmlUrl})` : + `**🚨BUILD FAILED🚨**: →→→?[${data.message}](${data.htmlUrl})←←←`; + + createPost({ text }, await asyncBot); +}; diff --git a/src/queue/processors/http/unfollow.ts b/src/queue/processors/http/unfollow.ts new file mode 100644 index 0000000000..d3d5f2246f --- /dev/null +++ b/src/queue/processors/http/unfollow.ts @@ -0,0 +1,56 @@ +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'; + +export default async ({ data }) => { + // Delete following + const following = await Following.findOneAndDelete({ _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 promisedPackedUser = packUser(followee, follower); + + if (isRemoteUser(followee)) { + const undo = renderUndo(renderFollow(follower, followee)); + undo['@context'] = context; + + await request(follower, followee.account.inbox, undo); + } + + stream(follower._id, 'unfollow', promisedPackedUser); + }) + ]); +}; diff --git a/src/queue/processors/index.ts b/src/queue/processors/index.ts new file mode 100644 index 0000000000..172048ddae --- /dev/null +++ b/src/queue/processors/index.ts @@ -0,0 +1,18 @@ +import queue from '../queue'; +import db from './db'; +import http from './http'; + +export default () => { + queue.process('db', db); + + /* + 256 is the default concurrency limit of Mozilla Firefox and Google + Chromium. + + a8af215e691f3a2205a3758d2d96e9d328e100ff - chromium/src.git - Git at Google + https://chromium.googlesource.com/chromium/src.git/+/a8af215e691f3a2205a3758d2d96e9d328e100ff + Network.http.max-connections - MozillaZine Knowledge Base + http://kb.mozillazine.org/Network.http.max-connections + */ + queue.process('http', 256, http); +}; -- cgit v1.2.3-freya From 5bd1451b610c134c056991e05327990180cbb8d5 Mon Sep 17 00:00:00 2001 From: syuilo Date: Wed, 4 Apr 2018 23:22:48 +0900 Subject: wip --- src/queue/processors/http/deliver.ts | 16 +++------------- 1 file changed, 3 insertions(+), 13 deletions(-) (limited to 'src/queue/processors') diff --git a/src/queue/processors/http/deliver.ts b/src/queue/processors/http/deliver.ts index 8cd9eb624e..1700063a5d 100644 --- a/src/queue/processors/http/deliver.ts +++ b/src/queue/processors/http/deliver.ts @@ -1,17 +1,7 @@ import * as kue from 'kue'; -import Channel from '../../models/channel'; -import Following from '../../models/following'; -import ChannelWatching from '../../models/channel-watching'; -import Post, { pack } from '../../models/post'; -import User, { isLocalUser } from '../../models/user'; -import stream, { publishChannelStream } from '../../publishers/stream'; -import context from '../../remote/activitypub/renderer/context'; -import renderCreate from '../../remote/activitypub/renderer/create'; -import renderNote from '../../remote/activitypub/renderer/note'; -import request from '../../remote/request'; +import request from '../../../remote/request'; export default async (job: kue.Job, done): Promise => { - - request(user, following.follower[0].account.inbox, create); -} + await request(job.data.user, job.data.to, job.data.content); +}; -- cgit v1.2.3-freya 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/queue/processors') 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 From 0ac695edbda10231ee70f0a0debedb4c8f708954 Mon Sep 17 00:00:00 2001 From: syuilo Date: Thu, 5 Apr 2018 00:01:01 +0900 Subject: wip --- src/queue/processors/http/follow.ts | 69 ------------------------------------- 1 file changed, 69 deletions(-) delete mode 100644 src/queue/processors/http/follow.ts (limited to 'src/queue/processors') diff --git a/src/queue/processors/http/follow.ts b/src/queue/processors/http/follow.ts deleted file mode 100644 index 8bf890efbc..0000000000 --- a/src/queue/processors/http/follow.ts +++ /dev/null @@ -1,69 +0,0 @@ -import User, { isLocalUser, pack as packUser } 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 render from '../../remote/activitypub/renderer/follow'; -import request from '../../remote/request'; - -export default ({ data }) => Following.findOne({ _id: data.following }).then(({ followerId, followeeId }) => { - const promisedFollower = User.findOne({ _id: followerId }); - const promisedFollowee = User.findOne({ _id: followeeId }); - - return Promise.all([ - // Increment following count - User.update(followerId, { - $inc: { - followingCount: 1 - } - }), - - promisedFollower.then(({ followingCount }) => FollowingLog.insert({ - createdAt: data.following.createdAt, - userId: followerId, - count: followingCount + 1 - })), - - // Increment followers count - User.update({ _id: followeeId }, { - $inc: { - followersCount: 1 - } - }), - - promisedFollowee.then(({ followersCount }) => FollowedLog.insert({ - createdAt: data.following.createdAt, - userId: followerId, - count: followersCount + 1 - })), - - // Notify - promisedFollowee.then(followee => followee.host === null ? - notify(followeeId, followerId, 'follow') : null), - - // Publish follow event - Promise.all([promisedFollower, promisedFollowee]).then(([follower, followee]) => { - let followerEvent; - let followeeEvent; - - if (isLocalUser(follower)) { - followerEvent = packUser(followee, follower) - .then(packed => event(follower._id, 'follow', packed)); - } - - if (isLocalUser(followee)) { - followeeEvent = packUser(follower, followee) - .then(packed => event(followee._id, 'followed', packed)); - } else if (isLocalUser(follower)) { - const rendered = render(follower, followee); - rendered['@context'] = context; - - followeeEvent = request(follower, followee.account.inbox, rendered); - } - - return Promise.all([followerEvent, followeeEvent]); - }) - ]); -}); -- cgit v1.2.3-freya From fba46b4c7f5751cd31463e922edb19f5652ca408 Mon Sep 17 00:00:00 2001 From: syuilo Date: Thu, 5 Apr 2018 01:07:07 +0900 Subject: wip --- src/queue/processors/http/index.ts | 10 +--- src/queue/processors/http/perform-activitypub.ts | 7 --- src/remote/activitypub/resolve-person.ts | 58 ++++++------------------ 3 files changed, 15 insertions(+), 60 deletions(-) delete mode 100644 src/queue/processors/http/perform-activitypub.ts (limited to 'src/queue/processors') diff --git a/src/queue/processors/http/index.ts b/src/queue/processors/http/index.ts index 8f9aa717c3..06c6b1d1aa 100644 --- a/src/queue/processors/http/index.ts +++ b/src/queue/processors/http/index.ts @@ -1,17 +1,11 @@ -import deliverPost from './deliver-post'; -import follow from './follow'; -import performActivityPub from './perform-activitypub'; +import deliver from './deliver'; import processInbox from './process-inbox'; import reportGitHubFailure from './report-github-failure'; -import unfollow from './unfollow'; const handlers = { - deliverPost, - follow, - performActivityPub, + deliver, processInbox, reportGitHubFailure, - unfollow }; export default (job, done) => handlers[job.data.type](job).then(() => done(), done); diff --git a/src/queue/processors/http/perform-activitypub.ts b/src/queue/processors/http/perform-activitypub.ts deleted file mode 100644 index 963e532fe5..0000000000 --- a/src/queue/processors/http/perform-activitypub.ts +++ /dev/null @@ -1,7 +0,0 @@ -import User from '../../models/user'; -import act from '../../remote/activitypub/act'; -import Resolver from '../../remote/activitypub/resolver'; - -export default ({ data }) => User.findOne({ _id: data.actor }) - .then(actor => act(new Resolver(), actor, data.outbox)) - .then(Promise.all); diff --git a/src/remote/activitypub/resolve-person.ts b/src/remote/activitypub/resolve-person.ts index 59be65908e..77d08398be 100644 --- a/src/remote/activitypub/resolve-person.ts +++ b/src/remote/activitypub/resolve-person.ts @@ -1,17 +1,15 @@ import { JSDOM } from 'jsdom'; import { toUnicode } from 'punycode'; import User, { validateUsername, isValidName, isValidDescription } from '../../models/user'; -import queue from '../../queue'; import webFinger from '../webfinger'; import create from './create'; import Resolver from './resolver'; - -async function isCollection(collection) { - return ['Collection', 'OrderedCollection'].includes(collection.type); -} +import uploadFromUrl from '../../api/drive/upload-from-url'; export default async (value, verifier?: string) => { - const { resolver, object } = await new Resolver().resolveOne(value); + const resolver = new Resolver(); + + const object = await resolver.resolve(value) as any; if ( object === null || @@ -21,24 +19,10 @@ export default async (value, verifier?: string) => { !isValidName(object.name) || !isValidDescription(object.summary) ) { - throw new Error(); + throw new Error('invalid person'); } - const [followers, following, outbox, finger] = await Promise.all([ - resolver.resolveOne(object.followers).then( - resolved => isCollection(resolved.object) ? resolved.object : null, - () => null - ), - resolver.resolveOne(object.following).then( - resolved => isCollection(resolved.object) ? resolved.object : null, - () => null - ), - resolver.resolveOne(object.outbox).then( - resolved => isCollection(resolved.object) ? resolved.object : null, - () => null - ), - webFinger(object.id, verifier), - ]); + const finger = await webFinger(object.id, verifier); const host = toUnicode(finger.subject.replace(/^.*?@/, '')); const hostLower = host.replace(/[A-Z]+/, matched => matched.toLowerCase()); @@ -50,10 +34,10 @@ export default async (value, verifier?: string) => { bannerId: null, createdAt: Date.parse(object.published), description: summaryDOM.textContent, - followersCount: followers ? followers.totalItem || 0 : 0, - followingCount: following ? following.totalItem || 0 : 0, + followersCount: 0, + followingCount: 0, name: object.name, - postsCount: outbox ? outbox.totalItem || 0 : 0, + postsCount: 0, driveCapacity: 1024 * 1024 * 8, // 8MiB username: object.preferredUsername, usernameLower: object.preferredUsername.toLowerCase(), @@ -69,33 +53,17 @@ export default async (value, verifier?: string) => { }, }); - queue.create('http', { - type: 'performActivityPub', - actor: user._id, - outbox - }).save(); - const [avatarId, bannerId] = await Promise.all([ object.icon, object.image - ].map(async value => { - if (value === undefined) { + ].map(async url => { + if (url === undefined) { return null; } - try { - const created = await create(resolver, user, value); + const img = await uploadFromUrl(url, user); - await Promise.all(created.map(asyncCreated => asyncCreated.then(created => { - if (created !== null && created.object.$ref === 'driveFiles.files') { - throw created.object.$id; - } - }, () => {}))); - - return null; - } catch (id) { - return id; - } + return img._id; })); User.update({ _id: user._id }, { $set: { avatarId, bannerId } }); -- cgit v1.2.3-freya From eb304cb5fb588a3da8742f234cdf05ce6deeaa59 Mon Sep 17 00:00:00 2001 From: syuilo Date: Thu, 5 Apr 2018 01:22:41 +0900 Subject: wip --- src/api/following/delete.ts | 69 +++++++++++++++++++++++++++++ src/queue/processors/http/unfollow.ts | 63 -------------------------- src/remote/activitypub/act/create.ts | 1 - src/remote/activitypub/act/index.ts | 11 +++-- src/remote/activitypub/act/undo.ts | 15 +++++++ src/remote/activitypub/act/undo/index.ts | 27 ----------- src/remote/activitypub/act/undo/unfollow.ts | 11 ----- src/remote/activitypub/act/unfollow.ts | 25 +++++++++++ 8 files changed, 114 insertions(+), 108 deletions(-) create mode 100644 src/api/following/delete.ts delete mode 100644 src/queue/processors/http/unfollow.ts create mode 100644 src/remote/activitypub/act/undo.ts delete mode 100644 src/remote/activitypub/act/undo/index.ts delete mode 100644 src/remote/activitypub/act/undo/unfollow.ts create mode 100644 src/remote/activitypub/act/unfollow.ts (limited to 'src/queue/processors') diff --git a/src/api/following/delete.ts b/src/api/following/delete.ts new file mode 100644 index 0000000000..4cdff7ce1b --- /dev/null +++ b/src/api/following/delete.ts @@ -0,0 +1,69 @@ +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 { createHttp } 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; + + createHttp({ + type: 'deliver', + user: follower, + content, + to: followee.account.inbox + }).save(); + } +} diff --git a/src/queue/processors/http/unfollow.ts b/src/queue/processors/http/unfollow.ts deleted file mode 100644 index 801a3612a7..0000000000 --- a/src/queue/processors/http/unfollow.ts +++ /dev/null @@ -1,63 +0,0 @@ -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 }) => { - const following = await Following.findOne({ _id: data.id }); - if (following === null) { - return; - } - - const [follower, followee] = await Promise.all([ - User.findOne({ _id: following.followerId }), - User.findOne({ _id: following.followeeId }) - ]); - - if (isLocalUser(follower) && isRemoteUser(followee)) { - const undo = renderUndo(renderFollow(follower, followee)); - undo['@context'] = context; - - 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); - - // 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 957900900f..c486571fc1 100644 --- a/src/remote/activitypub/act/create.ts +++ b/src/remote/activitypub/act/create.ts @@ -1,5 +1,4 @@ import { JSDOM } from 'jsdom'; -const createDOMPurify = require('dompurify'); import Resolver from '../resolver'; import DriveFile from '../../../models/drive-file'; diff --git a/src/remote/activitypub/act/index.ts b/src/remote/activitypub/act/index.ts index d78335f16e..f22500acef 100644 --- a/src/remote/activitypub/act/index.ts +++ b/src/remote/activitypub/act/index.ts @@ -2,25 +2,24 @@ import create from './create'; import performDeleteActivity from './delete'; import follow from './follow'; import undo from './undo'; -import Resolver from '../resolver'; import { IObject } from '../type'; -export default async (parentResolver: Resolver, actor, activity: IObject): Promise => { +export default async (actor, activity: IObject): Promise => { switch (activity.type) { case 'Create': - await create(parentResolver, actor, activity); + await create(actor, activity); break; case 'Delete': - await performDeleteActivity(parentResolver, actor, activity); + await performDeleteActivity(actor, activity); break; case 'Follow': - await follow(parentResolver, actor, activity); + await follow(actor, activity); break; case 'Undo': - await undo(parentResolver, actor, activity); + await undo(actor, activity); break; default: diff --git a/src/remote/activitypub/act/undo.ts b/src/remote/activitypub/act/undo.ts new file mode 100644 index 0000000000..b3b83777d1 --- /dev/null +++ b/src/remote/activitypub/act/undo.ts @@ -0,0 +1,15 @@ +import unfollow from './unfollow'; + +export default async (actor, activity): Promise => { + if ('actor' in activity && actor.account.uri !== activity.actor) { + throw new Error('invalid actor'); + } + + switch (activity.object.type) { + case 'Follow': + unfollow(activity.object); + break; + } + + return null; +}; diff --git a/src/remote/activitypub/act/undo/index.ts b/src/remote/activitypub/act/undo/index.ts deleted file mode 100644 index aa60d3a4fa..0000000000 --- a/src/remote/activitypub/act/undo/index.ts +++ /dev/null @@ -1,27 +0,0 @@ -import act from '../../act'; -import deleteObject from '../../delete'; -import unfollow from './unfollow'; -import Resolver from '../../resolver'; - -export default async (resolver: Resolver, actor, activity): Promise => { - if ('actor' in activity && actor.account.uri !== activity.actor) { - throw new Error(); - } - - const results = await act(resolver, actor, activity.object); - - await Promise.all(results.map(async promisedResult => { - const result = await promisedResult; - - if (result === null || await deleteObject(result) !== null) { - return; - } - - switch (result.object.$ref) { - case 'following': - await unfollow(result.object); - } - })); - - return null; -}; diff --git a/src/remote/activitypub/act/undo/unfollow.ts b/src/remote/activitypub/act/undo/unfollow.ts deleted file mode 100644 index c17e06e8a9..0000000000 --- a/src/remote/activitypub/act/undo/unfollow.ts +++ /dev/null @@ -1,11 +0,0 @@ -import queue from '../../../../queue'; - -export default ({ $id }) => new Promise((resolve, reject) => { - queue.create('http', { type: 'unfollow', id: $id }).save(error => { - if (error) { - reject(error); - } else { - resolve(); - } - }); -}); diff --git a/src/remote/activitypub/act/unfollow.ts b/src/remote/activitypub/act/unfollow.ts new file mode 100644 index 0000000000..e3c9e1c1c8 --- /dev/null +++ b/src/remote/activitypub/act/unfollow.ts @@ -0,0 +1,25 @@ +import parseAcct from '../../../acct/parse'; +import User from '../../../models/user'; +import config from '../../../config'; +import unfollow from '../../../api/following/delete'; + +export default async (actor, activity): Promise => { + const prefix = config.url + '/@'; + const id = activity.object.id || activity.object; + + if (!id.startsWith(prefix)) { + return null; + } + + const { username, host } = parseAcct(id.slice(prefix.length)); + if (host !== null) { + throw new Error(); + } + + const followee = await User.findOne({ username, host }); + if (followee === null) { + throw new Error(); + } + + await unfollow(actor, followee, activity); +}; -- cgit v1.2.3-freya From b6b98752053042b3b37d5b94bc823635edc2061f Mon Sep 17 00:00:00 2001 From: syuilo Date: Thu, 5 Apr 2018 01:24:01 +0900 Subject: wip --- src/queue/processors/http/process-inbox.ts | 12 ++++++------ src/queue/processors/index.ts | 18 ------------------ 2 files changed, 6 insertions(+), 24 deletions(-) delete mode 100644 src/queue/processors/index.ts (limited to 'src/queue/processors') diff --git a/src/queue/processors/http/process-inbox.ts b/src/queue/processors/http/process-inbox.ts index fff1fbf663..82585d3a6b 100644 --- a/src/queue/processors/http/process-inbox.ts +++ b/src/queue/processors/http/process-inbox.ts @@ -1,11 +1,11 @@ import * as kue from 'kue'; import { verifySignature } from 'http-signature'; -import parseAcct from '../../acct/parse'; -import User, { IRemoteUser } from '../../models/user'; -import act from '../../remote/activitypub/act'; -import resolvePerson from '../../remote/activitypub/resolve-person'; -import Resolver from '../../remote/activitypub/resolver'; +import parseAcct from '../../../acct/parse'; +import User, { IRemoteUser } from '../../../models/user'; +import act from '../../../remote/activitypub/act'; +import resolvePerson from '../../../remote/activitypub/resolve-person'; +import Resolver from '../../../remote/activitypub/resolver'; // ユーザーのinboxにアクティビティが届いた時の処理 export default async (job: kue.Job, done): Promise => { @@ -47,7 +47,7 @@ export default async (job: kue.Job, done): Promise => { // アクティビティを処理 try { - await act(new Resolver(), user, activity); + await act(user, activity); done(); } catch (e) { done(e); diff --git a/src/queue/processors/index.ts b/src/queue/processors/index.ts deleted file mode 100644 index 172048ddae..0000000000 --- a/src/queue/processors/index.ts +++ /dev/null @@ -1,18 +0,0 @@ -import queue from '../queue'; -import db from './db'; -import http from './http'; - -export default () => { - queue.process('db', db); - - /* - 256 is the default concurrency limit of Mozilla Firefox and Google - Chromium. - - a8af215e691f3a2205a3758d2d96e9d328e100ff - chromium/src.git - Git at Google - https://chromium.googlesource.com/chromium/src.git/+/a8af215e691f3a2205a3758d2d96e9d328e100ff - Network.http.max-connections - MozillaZine Knowledge Base - http://kb.mozillazine.org/Network.http.max-connections - */ - queue.process('http', 256, http); -}; -- cgit v1.2.3-freya From c2c03a1c650c0395fc4d77334ec86d00b1bb24c2 Mon Sep 17 00:00:00 2001 From: syuilo Date: Thu, 5 Apr 2018 03:21:11 +0900 Subject: wip --- src/api/drive/add-file.ts | 12 ++++----- src/api/drive/upload-from-url.ts | 2 +- src/api/post/create.ts | 13 +++++----- src/api/post/watch.ts | 2 +- src/queue/processors/db/delete-post-dependents.ts | 12 ++++----- src/queue/processors/http/process-inbox.ts | 1 - src/remote/activitypub/act/create.ts | 25 ++++++++++++------ src/remote/activitypub/act/delete.ts | 31 ++++++++++++++--------- src/remote/activitypub/act/undo.ts | 2 +- src/remote/activitypub/delete/index.ts | 10 -------- src/remote/activitypub/delete/post.ts | 13 ---------- src/remote/activitypub/resolve-person.ts | 1 - src/server/activitypub/inbox.ts | 4 +-- 13 files changed, 60 insertions(+), 68 deletions(-) delete mode 100644 src/remote/activitypub/delete/index.ts delete mode 100644 src/remote/activitypub/delete/post.ts (limited to 'src/queue/processors') diff --git a/src/api/drive/add-file.ts b/src/api/drive/add-file.ts index 24eb5208d5..64a2f18340 100644 --- a/src/api/drive/add-file.ts +++ b/src/api/drive/add-file.ts @@ -10,12 +10,12 @@ 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'; +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 diff --git a/src/api/drive/upload-from-url.ts b/src/api/drive/upload-from-url.ts index f96af0f266..26c890d157 100644 --- a/src/api/drive/upload-from-url.ts +++ b/src/api/drive/upload-from-url.ts @@ -1,5 +1,5 @@ import * as URL from 'url'; -import { IDriveFile, validateFileName } from '../models/drive-file'; +import { IDriveFile, validateFileName } from '../../models/drive-file'; import create from './add-file'; import * as debug from 'debug'; import * as tmp from 'tmp'; diff --git a/src/api/post/create.ts b/src/api/post/create.ts index 8256cbc355..36819ec2b8 100644 --- a/src/api/post/create.ts +++ b/src/api/post/create.ts @@ -1,6 +1,5 @@ -import parseAcct from '../../acct/parse'; import Post, { pack, IPost } from '../../models/post'; -import User, { isLocalUser, isRemoteUser, IUser } from '../../models/user'; +import User, { isLocalUser, IUser } from '../../models/user'; import stream from '../../publishers/stream'; import Following from '../../models/following'; import { createHttp } from '../../queue'; @@ -25,14 +24,16 @@ export default async (user: IUser, content: { repost: IPost; media: IDriveFile[]; geo: any; - poll: any; + poll?: any; viaMobile: boolean; - tags: string[]; - cw: string; - visibility: string; + tags?: string[]; + cw?: string; + visibility?: string; uri?: string; app?: IApp; }) => new Promise(async (res, rej) => { + if (content.visibility == null) content.visibility = 'public'; + const tags = content.tags || []; let tokens = null; diff --git a/src/api/post/watch.ts b/src/api/post/watch.ts index 61ea444430..bbd9976f40 100644 --- a/src/api/post/watch.ts +++ b/src/api/post/watch.ts @@ -1,5 +1,5 @@ import * as mongodb from 'mongodb'; -import Watching from '../models/post-watching'; +import Watching from '../../models/post-watching'; export default async (me: mongodb.ObjectID, post: object) => { // 自分の投稿はwatchできない diff --git a/src/queue/processors/db/delete-post-dependents.ts b/src/queue/processors/db/delete-post-dependents.ts index 879c41ec9c..6de21eb053 100644 --- a/src/queue/processors/db/delete-post-dependents.ts +++ b/src/queue/processors/db/delete-post-dependents.ts @@ -1,9 +1,9 @@ -import Favorite from '../../models/favorite'; -import Notification from '../../models/notification'; -import PollVote from '../../models/poll-vote'; -import PostReaction from '../../models/post-reaction'; -import PostWatching from '../../models/post-watching'; -import Post from '../../models/post'; +import Favorite from '../../../models/favorite'; +import Notification from '../../../models/notification'; +import PollVote from '../../../models/poll-vote'; +import PostReaction from '../../../models/post-reaction'; +import PostWatching from '../../../models/post-watching'; +import Post from '../../../models/post'; export default async ({ data }) => Promise.all([ Favorite.remove({ postId: data._id }), diff --git a/src/queue/processors/http/process-inbox.ts b/src/queue/processors/http/process-inbox.ts index 82585d3a6b..c3074429f8 100644 --- a/src/queue/processors/http/process-inbox.ts +++ b/src/queue/processors/http/process-inbox.ts @@ -5,7 +5,6 @@ import parseAcct from '../../../acct/parse'; import User, { IRemoteUser } from '../../../models/user'; import act from '../../../remote/activitypub/act'; import resolvePerson from '../../../remote/activitypub/resolve-person'; -import Resolver from '../../../remote/activitypub/resolver'; // ユーザーのinboxにアクティビティが届いた時の処理 export default async (job: kue.Job, done): Promise => { diff --git a/src/remote/activitypub/act/create.ts b/src/remote/activitypub/act/create.ts index c486571fc1..f97832a989 100644 --- a/src/remote/activitypub/act/create.ts +++ b/src/remote/activitypub/act/create.ts @@ -36,17 +36,17 @@ export default async (actor, activity): Promise => { switch (object.type) { case 'Image': - createImage(resolver, object); + createImage(object); break; case 'Note': - createNote(resolver, object); + createNote(object); break; } /// - async function createImage(resolver: Resolver, image) { + async function createImage(image) { if ('attributedTo' in image && actor.account.uri !== image.attributedTo) { throw new Error('invalid image'); } @@ -54,7 +54,7 @@ export default async (actor, activity): Promise => { return await uploadFromUrl(image.url, actor); } - async function createNote(resolver: Resolver, note) { + async function createNote(note) { if ( ('attributedTo' in note && actor.account.uri !== note.attributedTo) || typeof note.id !== 'string' @@ -63,20 +63,29 @@ export default async (actor, activity): Promise => { } const media = []; - if ('attachment' in note) { note.attachment.forEach(async media => { - const created = await createImage(resolver, media); + const created = await createImage(media); media.push(created); }); } + let reply = null; + if ('inReplyTo' in note) { + const inReplyToPost = await Post.findOne({ uri: note.id || note }); + if (inReplyToPost) { + reply = inReplyToPost; + } else { + reply = await createNote(await resolver.resolve(note)); + } + } + const { window } = new JSDOM(note.content); - await createPost(actor, { + return await createPost(actor, { createdAt: new Date(note.published), media, - reply: undefined, + reply, repost: undefined, text: window.document.body.textContent, viaMobile: false, diff --git a/src/remote/activitypub/act/delete.ts b/src/remote/activitypub/act/delete.ts index f9eb4dd08d..334ca47edf 100644 --- a/src/remote/activitypub/act/delete.ts +++ b/src/remote/activitypub/act/delete.ts @@ -1,21 +1,28 @@ -import create from '../create'; -import deleteObject from '../delete'; +import Resolver from '../resolver'; +import Post from '../../../models/post'; +import { createDb } from '../../../queue'; -export default async (resolver, actor, activity) => { +export default async (actor, activity): Promise => { if ('actor' in activity && actor.account.uri !== activity.actor) { throw new Error(); } - const results = await create(resolver, actor, activity.object); + const resolver = new Resolver(); - await Promise.all(results.map(async promisedResult => { - const result = await promisedResult; - if (result === null) { - return; - } + const object = await resolver.resolve(activity); - await deleteObject(result); - })); + switch (object.type) { + case 'Note': + deleteNote(object); + break; + } + + async function deleteNote(note) { + const post = await Post.findOneAndDelete({ uri: note.id }); - return null; + createDb({ + type: 'deletePostDependents', + id: post._id + }).delay(65536).save(); + } }; diff --git a/src/remote/activitypub/act/undo.ts b/src/remote/activitypub/act/undo.ts index b3b83777d1..9d9f6b0359 100644 --- a/src/remote/activitypub/act/undo.ts +++ b/src/remote/activitypub/act/undo.ts @@ -7,7 +7,7 @@ export default async (actor, activity): Promise => { switch (activity.object.type) { case 'Follow': - unfollow(activity.object); + unfollow(actor, activity.object); break; } diff --git a/src/remote/activitypub/delete/index.ts b/src/remote/activitypub/delete/index.ts deleted file mode 100644 index bc9104284b..0000000000 --- a/src/remote/activitypub/delete/index.ts +++ /dev/null @@ -1,10 +0,0 @@ -import deletePost from './post'; - -export default async ({ object }) => { - switch (object.$ref) { - case 'posts': - return deletePost(object); - } - - return null; -}; diff --git a/src/remote/activitypub/delete/post.ts b/src/remote/activitypub/delete/post.ts deleted file mode 100644 index f6c816647d..0000000000 --- a/src/remote/activitypub/delete/post.ts +++ /dev/null @@ -1,13 +0,0 @@ -import Post from '../../../models/post'; -import queue from '../../../queue'; - -export default async ({ $id }) => { - const promisedDeletion = Post.findOneAndDelete({ _id: $id }); - - await new Promise((resolve, reject) => queue.create('db', { - type: 'deletePostDependents', - id: $id - }).delay(65536).save(error => error ? reject(error) : resolve())); - - return promisedDeletion; -}; diff --git a/src/remote/activitypub/resolve-person.ts b/src/remote/activitypub/resolve-person.ts index 77d08398be..28162497f3 100644 --- a/src/remote/activitypub/resolve-person.ts +++ b/src/remote/activitypub/resolve-person.ts @@ -2,7 +2,6 @@ import { JSDOM } from 'jsdom'; import { toUnicode } from 'punycode'; import User, { validateUsername, isValidName, isValidDescription } from '../../models/user'; import webFinger from '../webfinger'; -import create from './create'; import Resolver from './resolver'; import uploadFromUrl from '../../api/drive/upload-from-url'; diff --git a/src/server/activitypub/inbox.ts b/src/server/activitypub/inbox.ts index 847dc19af6..b0015409a9 100644 --- a/src/server/activitypub/inbox.ts +++ b/src/server/activitypub/inbox.ts @@ -1,7 +1,7 @@ import * as bodyParser from 'body-parser'; import * as express from 'express'; import { parseRequest } from 'http-signature'; -import queue from '../../queue'; +import { createHttp } from '../../queue'; const app = express(); @@ -22,7 +22,7 @@ app.post('/@:user/inbox', bodyParser.json({ return res.sendStatus(401); } - queue.create('http', { + createHttp({ type: 'processInbox', activity: req.body, signature, -- cgit v1.2.3-freya From 7403f38fb43b0ad747236061a591cbf94e198ba6 Mon Sep 17 00:00:00 2001 From: syuilo Date: Thu, 5 Apr 2018 18:43:06 +0900 Subject: wip --- src/api/post/create.ts | 15 ++++++++------- src/index.ts | 2 +- src/queue/index.ts | 8 +++++++- src/queue/processors/http/deliver.ts | 9 ++++++++- src/queue/processors/http/index.ts | 16 ++++++++++++---- src/queue/processors/http/report-github-failure.ts | 6 +++--- src/remote/activitypub/resolver.ts | 2 +- src/remote/request.ts | 8 ++++++++ src/server/api/endpoints/following/create.ts | 11 ++--------- 9 files changed, 50 insertions(+), 27 deletions(-) (limited to 'src/queue/processors') diff --git a/src/api/post/create.ts b/src/api/post/create.ts index 5495117538..dbeb87ae86 100644 --- a/src/api/post/create.ts +++ b/src/api/post/create.ts @@ -18,20 +18,21 @@ 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; + createdAt?: Date; + text?: string; + reply?: IPost; + repost?: IPost; + media?: IDriveFile[]; + geo?: any; poll?: any; - viaMobile: boolean; + viaMobile?: boolean; tags?: string[]; cw?: string; visibility?: string; uri?: string; app?: IApp; }) => new Promise(async (res, rej) => { + if (content.createdAt == null) content.createdAt = new Date(); if (content.visibility == null) content.visibility = 'public'; const tags = content.tags || []; diff --git a/src/index.ts b/src/index.ts index e35c917a44..f45bcaa6ad 100644 --- a/src/index.ts +++ b/src/index.ts @@ -103,7 +103,7 @@ async function workerMain(opt) { if (!opt['only-server']) { // start processor - require('./processor').default(); + require('./queue').default(); } // Send a 'ready' message to parent process diff --git a/src/queue/index.ts b/src/queue/index.ts index c8c436b18c..86600dc265 100644 --- a/src/queue/index.ts +++ b/src/queue/index.ts @@ -1,8 +1,12 @@ import { createQueue } from 'kue'; +import * as debug from 'debug'; + import config from '../config'; import db from './processors/db'; import http from './processors/http'; +const log = debug('misskey:queue'); + const queue = createQueue({ redis: { port: config.redis.port, @@ -12,6 +16,8 @@ const queue = createQueue({ }); export function createHttp(data) { + log(`HTTP job created: ${JSON.stringify(data)}`); + return queue .create('http', data) .attempts(16) @@ -22,7 +28,7 @@ export function createDb(data) { return queue.create('db', data); } -export function process() { +export default function() { queue.process('db', db); /* diff --git a/src/queue/processors/http/deliver.ts b/src/queue/processors/http/deliver.ts index 1700063a5d..da7e8bc368 100644 --- a/src/queue/processors/http/deliver.ts +++ b/src/queue/processors/http/deliver.ts @@ -3,5 +3,12 @@ import * as kue from 'kue'; import request from '../../../remote/request'; export default async (job: kue.Job, done): Promise => { - await request(job.data.user, job.data.to, job.data.content); + try { + await request(job.data.user, job.data.to, job.data.content); + done(); + } catch (e) { + console.warn(`deliver failed: ${e}`); + + done(e); + } }; diff --git a/src/queue/processors/http/index.ts b/src/queue/processors/http/index.ts index 06c6b1d1aa..3d7d941b1a 100644 --- a/src/queue/processors/http/index.ts +++ b/src/queue/processors/http/index.ts @@ -3,9 +3,17 @@ import processInbox from './process-inbox'; import reportGitHubFailure from './report-github-failure'; const handlers = { - deliver, - processInbox, - reportGitHubFailure, + deliver, + processInbox, + reportGitHubFailure }; -export default (job, done) => handlers[job.data.type](job).then(() => done(), done); +export default (job, done) => { + const handler = handlers[job.data.type]; + + if (handler) { + handler(job).then(() => done(), done); + } else { + console.warn(`Unknown job: ${job.data.type}`); + } +}; diff --git a/src/queue/processors/http/report-github-failure.ts b/src/queue/processors/http/report-github-failure.ts index 4f6f5ccee5..e747d062d3 100644 --- a/src/queue/processors/http/report-github-failure.ts +++ b/src/queue/processors/http/report-github-failure.ts @@ -1,6 +1,6 @@ import * as request from 'request-promise-native'; -import User from '../../models/user'; -const createPost = require('../../server/api/endpoints/posts/create'); +import User from '../../../models/user'; +import createPost from '../../../api/post/create'; export default async ({ data }) => { const asyncBot = User.findOne({ _id: data.userId }); @@ -20,5 +20,5 @@ export default async ({ data }) => { `**⚠️BUILD STILL FAILED⚠️**: ?[${data.message}](${data.htmlUrl})` : `**🚨BUILD FAILED🚨**: →→→?[${data.message}](${data.htmlUrl})←←←`; - createPost({ text }, await asyncBot); + createPost(await asyncBot, { text }); }; diff --git a/src/remote/activitypub/resolver.ts b/src/remote/activitypub/resolver.ts index 09a6e70056..38639c6813 100644 --- a/src/remote/activitypub/resolver.ts +++ b/src/remote/activitypub/resolver.ts @@ -59,7 +59,7 @@ export default class Resolver { throw new Error('invalid response'); } - log(`resolved: ${JSON.stringify(object)}`); + log(`resolved: ${JSON.stringify(object, null, 2)}`); return object; } diff --git a/src/remote/request.ts b/src/remote/request.ts index 72262cbf61..a375aebfbb 100644 --- a/src/remote/request.ts +++ b/src/remote/request.ts @@ -1,9 +1,15 @@ import { request } from 'https'; import { sign } from 'http-signature'; import { URL } from 'url'; +import * as debug from 'debug'; + import config from '../config'; +const log = debug('misskey:activitypub:deliver'); + export default ({ account, username }, url, object) => new Promise((resolve, reject) => { + log(`--> ${url}`); + const { protocol, hostname, port, pathname, search } = new URL(url); const req = request({ @@ -14,6 +20,8 @@ export default ({ account, username }, url, object) => new Promise((resolve, rej path: pathname + search, }, res => { res.on('end', () => { + log(`${url} --> ${res.statusCode}`); + if (res.statusCode >= 200 && res.statusCode < 300) { resolve(); } else { diff --git a/src/server/api/endpoints/following/create.ts b/src/server/api/endpoints/following/create.ts index e568595215..fae686ce54 100644 --- a/src/server/api/endpoints/following/create.ts +++ b/src/server/api/endpoints/following/create.ts @@ -4,7 +4,7 @@ import $ from 'cafy'; import User from '../../../../models/user'; import Following from '../../../../models/following'; -import queue from '../../../../queue'; +import create from '../../../../api/following/create'; /** * Follow a user @@ -50,15 +50,8 @@ module.exports = (params, user) => new Promise(async (res, rej) => { } // Create following - const { _id } = await Following.insert({ - createdAt: new Date(), - followerId: follower._id, - followeeId: followee._id - }); - - queue.create('http', { type: 'follow', following: _id }).save(); + create(follower, followee); // Send response res(); - }); -- cgit v1.2.3-freya 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/api/drive/add-file.ts | 314 ------------------- src/api/drive/upload-from-url.ts | 52 ---- src/api/following/create.ts | 72 ----- src/api/following/delete.ts | 64 ---- src/api/post/create.ts | 344 --------------------- src/api/post/watch.ts | 26 -- src/queue/processors/http/report-github-failure.ts | 2 +- src/remote/activitypub/act/create.ts | 4 +- src/remote/activitypub/act/follow.ts | 2 +- src/remote/activitypub/act/unfollow.ts | 2 +- src/remote/activitypub/resolve-person.ts | 2 +- src/server/api/endpoints/following/create.ts | 2 +- src/server/api/endpoints/posts/create.ts | 2 +- 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 ++ 20 files changed, 880 insertions(+), 880 deletions(-) delete mode 100644 src/api/drive/add-file.ts delete mode 100644 src/api/drive/upload-from-url.ts delete mode 100644 src/api/following/create.ts delete mode 100644 src/api/following/delete.ts delete mode 100644 src/api/post/create.ts delete mode 100644 src/api/post/watch.ts 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/queue/processors') diff --git a/src/api/drive/add-file.ts b/src/api/drive/add-file.ts deleted file mode 100644 index 64a2f18340..0000000000 --- a/src/api/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/api/drive/upload-from-url.ts b/src/api/drive/upload-from-url.ts deleted file mode 100644 index 676586cd15..0000000000 --- a/src/api/drive/upload-from-url.ts +++ /dev/null @@ -1,52 +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: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/api/following/create.ts b/src/api/following/create.ts deleted file mode 100644 index d919f4487f..0000000000 --- a/src/api/following/create.ts +++ /dev/null @@ -1,72 +0,0 @@ -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/api/following/delete.ts b/src/api/following/delete.ts deleted file mode 100644 index 364a4803b9..0000000000 --- a/src/api/following/delete.ts +++ /dev/null @@ -1,64 +0,0 @@ -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/api/post/create.ts b/src/api/post/create.ts deleted file mode 100644 index 9723dbe452..0000000000 --- a/src/api/post/create.ts +++ /dev/null @@ -1,344 +0,0 @@ -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/api/post/watch.ts b/src/api/post/watch.ts deleted file mode 100644 index bbd9976f40..0000000000 --- a/src/api/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/report-github-failure.ts b/src/queue/processors/http/report-github-failure.ts index e747d062d3..1e0b51f89f 100644 --- a/src/queue/processors/http/report-github-failure.ts +++ b/src/queue/processors/http/report-github-failure.ts @@ -1,6 +1,6 @@ import * as request from 'request-promise-native'; import User from '../../../models/user'; -import createPost from '../../../api/post/create'; +import createPost from '../../../services/post/create'; export default async ({ data }) => { const asyncBot = User.findOne({ _id: data.userId }); diff --git a/src/remote/activitypub/act/create.ts b/src/remote/activitypub/act/create.ts index fe58f58f87..139c98f3b3 100644 --- a/src/remote/activitypub/act/create.ts +++ b/src/remote/activitypub/act/create.ts @@ -3,8 +3,8 @@ import * as debug from 'debug'; import Resolver from '../resolver'; import Post from '../../../models/post'; -import uploadFromUrl from '../../../api/drive/upload-from-url'; -import createPost from '../../../api/post/create'; +import uploadFromUrl from '../../../services/drive/upload-from-url'; +import createPost from '../../../services/post/create'; import { IRemoteUser, isRemoteUser } from '../../../models/user'; import resolvePerson from '../resolve-person'; diff --git a/src/remote/activitypub/act/follow.ts b/src/remote/activitypub/act/follow.ts index dc173a0acb..4fc423d15c 100644 --- a/src/remote/activitypub/act/follow.ts +++ b/src/remote/activitypub/act/follow.ts @@ -1,7 +1,7 @@ import parseAcct from '../../../acct/parse'; import User from '../../../models/user'; import config from '../../../config'; -import follow from '../../../api/following/create'; +import follow from '../../../services/following/create'; export default async (actor, activity): Promise => { const prefix = config.url + '/@'; diff --git a/src/remote/activitypub/act/unfollow.ts b/src/remote/activitypub/act/unfollow.ts index e3c9e1c1c8..66c15e9a91 100644 --- a/src/remote/activitypub/act/unfollow.ts +++ b/src/remote/activitypub/act/unfollow.ts @@ -1,7 +1,7 @@ import parseAcct from '../../../acct/parse'; import User from '../../../models/user'; import config from '../../../config'; -import unfollow from '../../../api/following/delete'; +import unfollow from '../../../services/following/delete'; export default async (actor, activity): Promise => { const prefix = config.url + '/@'; diff --git a/src/remote/activitypub/resolve-person.ts b/src/remote/activitypub/resolve-person.ts index 2bf7a13540..907f198342 100644 --- a/src/remote/activitypub/resolve-person.ts +++ b/src/remote/activitypub/resolve-person.ts @@ -3,7 +3,7 @@ import { toUnicode } from 'punycode'; import User, { validateUsername, isValidName, isValidDescription } from '../../models/user'; import webFinger from '../webfinger'; import Resolver from './resolver'; -import uploadFromUrl from '../../api/drive/upload-from-url'; +import uploadFromUrl from '../../services/drive/upload-from-url'; export default async (value, verifier?: string) => { const resolver = new Resolver(); diff --git a/src/server/api/endpoints/following/create.ts b/src/server/api/endpoints/following/create.ts index fae686ce54..0ccac8d83d 100644 --- a/src/server/api/endpoints/following/create.ts +++ b/src/server/api/endpoints/following/create.ts @@ -4,7 +4,7 @@ import $ from 'cafy'; import User from '../../../../models/user'; import Following from '../../../../models/following'; -import create from '../../../../api/following/create'; +import create from '../../../../services/following/create'; /** * Follow a user diff --git a/src/server/api/endpoints/posts/create.ts b/src/server/api/endpoints/posts/create.ts index d241c8c387..003a892bc0 100644 --- a/src/server/api/endpoints/posts/create.ts +++ b/src/server/api/endpoints/posts/create.ts @@ -7,7 +7,7 @@ import Post, { IPost, isValidText, isValidCw, pack } from '../../../../models/po import { ILocalUser } from '../../../../models/user'; import Channel, { IChannel } from '../../../../models/channel'; import DriveFile from '../../../../models/drive-file'; -import create from '../../../../api/post/create'; +import create from '../../../../services/post/create'; import { IApp } from '../../../../models/app'; /** 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/queue/processors') 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 862463a13cc952919322b4e2f06c196bf2850517 Mon Sep 17 00:00:00 2001 From: syuilo Date: Fri, 6 Apr 2018 14:35:17 +0900 Subject: Fix bug --- src/queue/processors/http/process-inbox.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) (limited to 'src/queue/processors') diff --git a/src/queue/processors/http/process-inbox.ts b/src/queue/processors/http/process-inbox.ts index c3074429f8..4666e7f373 100644 --- a/src/queue/processors/http/process-inbox.ts +++ b/src/queue/processors/http/process-inbox.ts @@ -19,6 +19,7 @@ export default async (job: kue.Job, done): Promise => { if (host === null) { console.warn(`request was made by local user: @${username}`); done(); + return; } user = await User.findOne({ usernameLower: username, hostLower: host }) as IRemoteUser; @@ -40,7 +41,8 @@ export default async (job: kue.Job, done): Promise => { } if (!verifySignature(signature, user.account.publicKey.publicKeyPem)) { - done(new Error('signature verification failed')); + console.warn('signature verification failed'); + done(); return; } -- cgit v1.2.3-freya From 0856f4cd12ba2e053ab3ef30acdbec7e9a59fc61 Mon Sep 17 00:00:00 2001 From: syuilo Date: Fri, 6 Apr 2018 14:39:28 +0900 Subject: Use error instaed of warn --- src/queue/processors/http/index.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'src/queue/processors') diff --git a/src/queue/processors/http/index.ts b/src/queue/processors/http/index.ts index 61d7f9ac94..3dc2595374 100644 --- a/src/queue/processors/http/index.ts +++ b/src/queue/processors/http/index.ts @@ -14,7 +14,7 @@ export default (job, done) => { if (handler) { handler(job, done); } else { - console.warn(`Unknown job: ${job.data.type}`); + console.error(`Unknown job: ${job.data.type}`); done(); } }; -- cgit v1.2.3-freya From 3cf6eab11945e5b2577ef67117a51a210529e456 Mon Sep 17 00:00:00 2001 From: syuilo Date: Fri, 6 Apr 2018 22:40:06 +0900 Subject: Log --- src/queue/processors/http/process-inbox.ts | 10 ++++++++++ 1 file changed, 10 insertions(+) (limited to 'src/queue/processors') diff --git a/src/queue/processors/http/process-inbox.ts b/src/queue/processors/http/process-inbox.ts index 4666e7f373..eb4b62d37f 100644 --- a/src/queue/processors/http/process-inbox.ts +++ b/src/queue/processors/http/process-inbox.ts @@ -1,4 +1,5 @@ import * as kue from 'kue'; +import * as debug from 'debug'; import { verifySignature } from 'http-signature'; import parseAcct from '../../../acct/parse'; @@ -6,11 +7,20 @@ import User, { IRemoteUser } from '../../../models/user'; import act from '../../../remote/activitypub/act'; import resolvePerson from '../../../remote/activitypub/resolve-person'; +const log = debug('misskey:queue:inbox'); + // ユーザーのinboxにアクティビティが届いた時の処理 export default async (job: kue.Job, done): Promise => { const signature = job.data.signature; const activity = job.data.activity; + //#region Log + const info = Object.assign({}, activity); + delete info['@context']; + delete info['signature']; + log(info); + //#endregion + const keyIdLower = signature.keyId.toLowerCase(); let user; -- cgit v1.2.3-freya From 66346495e56e07a5b167a8f94ed971f081c94e88 Mon Sep 17 00:00:00 2001 From: syuilo Date: Sat, 7 Apr 2018 06:22:03 +0900 Subject: Fix bug --- src/queue/processors/http/deliver.ts | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) (limited to 'src/queue/processors') diff --git a/src/queue/processors/http/deliver.ts b/src/queue/processors/http/deliver.ts index da7e8bc368..f5d162fd0c 100644 --- a/src/queue/processors/http/deliver.ts +++ b/src/queue/processors/http/deliver.ts @@ -6,9 +6,14 @@ export default async (job: kue.Job, done): Promise => { try { await request(job.data.user, job.data.to, job.data.content); done(); - } catch (e) { - console.warn(`deliver failed: ${e}`); - - done(e); + } catch (res) { + if (res.statusCode >= 300 && res.statusCode < 400) { + // HTTPステータスコード4xxはクライアントエラーであり、それはつまり + // 何回再送しても成功することはないということなのでエラーにはしないでおく + done(); + } else { + console.warn(`deliver failed: ${res.statusMessage}`); + done(new Error(res.statusMessage)); + } } }; -- cgit v1.2.3-freya From 866797a0058a0eac1e5c1d7f6cdeeab845a505ae Mon Sep 17 00:00:00 2001 From: syuilo Date: Sat, 7 Apr 2018 06:23:38 +0900 Subject: Fix bug --- src/queue/processors/http/deliver.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'src/queue/processors') diff --git a/src/queue/processors/http/deliver.ts b/src/queue/processors/http/deliver.ts index f5d162fd0c..422e355b5f 100644 --- a/src/queue/processors/http/deliver.ts +++ b/src/queue/processors/http/deliver.ts @@ -7,7 +7,7 @@ export default async (job: kue.Job, done): Promise => { await request(job.data.user, job.data.to, job.data.content); done(); } catch (res) { - if (res.statusCode >= 300 && res.statusCode < 400) { + if (res.statusCode >= 400 && res.statusCode < 500) { // HTTPステータスコード4xxはクライアントエラーであり、それはつまり // 何回再送しても成功することはないということなのでエラーにはしないでおく done(); -- cgit v1.2.3-freya From 494597236cea4a40bddd9655b6506464df053bfe Mon Sep 17 00:00:00 2001 From: syuilo Date: Sat, 7 Apr 2018 07:19:30 +0900 Subject: 投稿に関しては論理削除するように MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 処理をシンプルにするため --- src/models/post.ts | 1 + src/queue/index.ts | 7 ------- src/queue/processors/db/delete-post-dependents.ts | 22 ---------------------- src/queue/processors/db/index.ts | 7 ------- src/remote/activitypub/act/delete/note.ts | 16 +++++++++------- 5 files changed, 10 insertions(+), 43 deletions(-) delete mode 100644 src/queue/processors/db/delete-post-dependents.ts delete mode 100644 src/queue/processors/db/index.ts (limited to 'src/queue/processors') diff --git a/src/models/post.ts b/src/models/post.ts index 68a638fa2f..ac7890d2e6 100644 --- a/src/models/post.ts +++ b/src/models/post.ts @@ -27,6 +27,7 @@ export type IPost = { _id: mongo.ObjectID; channelId: mongo.ObjectID; createdAt: Date; + deletedAt: Date; mediaIds: mongo.ObjectID[]; replyId: mongo.ObjectID; repostId: mongo.ObjectID; diff --git a/src/queue/index.ts b/src/queue/index.ts index 691223de2d..4aa1dc032d 100644 --- a/src/queue/index.ts +++ b/src/queue/index.ts @@ -1,7 +1,6 @@ import { createQueue } from 'kue'; import config from '../config'; -import db from './processors/db'; import http from './processors/http'; const queue = createQueue({ @@ -19,10 +18,6 @@ export function createHttp(data) { .backoff({ delay: 16384, type: 'exponential' }); } -export function createDb(data) { - return queue.create('db', data); -} - export function deliver(user, content, to) { return createHttp({ type: 'deliver', @@ -33,8 +28,6 @@ export function deliver(user, content, to) { } export default function() { - queue.process('db', db); - /* 256 is the default concurrency limit of Mozilla Firefox and Google Chromium. diff --git a/src/queue/processors/db/delete-post-dependents.ts b/src/queue/processors/db/delete-post-dependents.ts deleted file mode 100644 index 6de21eb053..0000000000 --- a/src/queue/processors/db/delete-post-dependents.ts +++ /dev/null @@ -1,22 +0,0 @@ -import Favorite from '../../../models/favorite'; -import Notification from '../../../models/notification'; -import PollVote from '../../../models/poll-vote'; -import PostReaction from '../../../models/post-reaction'; -import PostWatching from '../../../models/post-watching'; -import Post from '../../../models/post'; - -export default async ({ data }) => Promise.all([ - Favorite.remove({ postId: data._id }), - Notification.remove({ postId: data._id }), - PollVote.remove({ postId: data._id }), - PostReaction.remove({ postId: data._id }), - PostWatching.remove({ postId: data._id }), - Post.find({ repostId: data._id }).then(reposts => Promise.all([ - Notification.remove({ - postId: { - $in: reposts.map(({ _id }) => _id) - } - }), - Post.remove({ repostId: data._id }) - ])) -]); diff --git a/src/queue/processors/db/index.ts b/src/queue/processors/db/index.ts deleted file mode 100644 index 75838c099b..0000000000 --- a/src/queue/processors/db/index.ts +++ /dev/null @@ -1,7 +0,0 @@ -import deletePostDependents from './delete-post-dependents'; - -const handlers = { - deletePostDependents -}; - -export default (job, done) => handlers[job.data.type](job).then(() => done(), done); diff --git a/src/remote/activitypub/act/delete/note.ts b/src/remote/activitypub/act/delete/note.ts index ff9a8ee5f8..8e9447b481 100644 --- a/src/remote/activitypub/act/delete/note.ts +++ b/src/remote/activitypub/act/delete/note.ts @@ -1,7 +1,6 @@ import * as debug from 'debug'; import Post from '../../../../models/post'; -import { createDb } from '../../../../queue'; import { IRemoteUser } from '../../../../models/user'; const log = debug('misskey:activitypub'); @@ -19,10 +18,13 @@ export default async function(actor: IRemoteUser, uri: string): Promise { throw new Error('投稿を削除しようとしているユーザーは投稿の作成者ではありません'); } - Post.remove({ _id: post._id }); - - createDb({ - type: 'deletePostDependents', - id: post._id - }).delay(65536).save(); + Post.update({ _id: post._id }, { + $set: { + deletedAt: new Date(), + text: null, + textHtml: null, + mediaIds: [], + poll: null + } + }); } -- cgit v1.2.3-freya From c77013ab3ec748ccc441c497cd04231398194cc6 Mon Sep 17 00:00:00 2001 From: syuilo Date: Sat, 7 Apr 2018 15:45:03 +0900 Subject: oops --- src/queue/processors/http/index.ts | 25 ++++++++++++++----------- 1 file changed, 14 insertions(+), 11 deletions(-) (limited to 'src/queue/processors') diff --git a/src/queue/processors/http/index.ts b/src/queue/processors/http/index.ts index 0ea79305c6..3dc2595374 100644 --- a/src/queue/processors/http/index.ts +++ b/src/queue/processors/http/index.ts @@ -1,17 +1,20 @@ -import deliverPost from './deliver-post'; -import follow from './follow'; -import performActivityPub from './perform-activitypub'; +import deliver from './deliver'; import processInbox from './process-inbox'; import reportGitHubFailure from './report-github-failure'; -import unfollow from './unfollow'; const handlers = { - deliverPost, - follow, - performActivityPub, - processInbox, - reportGitHubFailure, - unfollow + deliver, + processInbox, + reportGitHubFailure }; -export default (job, done) => handlers[job.data.type](job, done); +export default (job, done) => { + const handler = handlers[job.data.type]; + + if (handler) { + handler(job, done); + } else { + console.error(`Unknown job: ${job.data.type}`); + done(); + } +}; -- cgit v1.2.3-freya