Retry HTTP requests

src/following/distribute.ts

+import User, { pack as packUser } from '../models/user';
+import FollowingLog from '../models/following-log';
+import FollowedLog from '../models/followed-log';
+import event from '../publishers/stream';
+import notify from '../publishers/notify';
+
+export default async (follower, followee) => Promise.all([
+	// Increment following count
+	User.update(follower._id, {
+		$inc: {
+			followingCount: 1
+		}
+	}),
+
+	FollowingLog.insert({
+		createdAt: new Date(),
+		userId: followee._id,
+		count: follower.followingCount + 1
+	}),
+
+	// Increment followers count
+	User.update({ _id: followee._id }, {
+		$inc: {
+			followersCount: 1
+		}
+	}),
+
+	FollowedLog.insert({
+		createdAt: new Date(),
+		userId: follower._id,
+		count: followee.followersCount + 1
+	}),
+
+	followee.host === null && Promise.all([
+		// Notify
+		notify(followee.id, follower.id, 'follow'),
+
+		// Publish follow event
+		packUser(follower, followee)
+			.then(packed => event(followee._id, 'followed', packed))
+	])
+]);

src/index.ts

@@ -99,7 +99,7 @@ async function workerMain(opt) {
 
 	if (!opt['only-server']) {
 		// start processor
-		require('./processor').default();
+		require('./queue').process();
 	}
 
 	// Send a 'ready' message to parent process

src/post/distribute.ts

@@ -8,7 +8,7 @@ import User, { isLocalUser } from '../models/user';
 import stream, { publishChannelStream } from '../publishers/stream';
 import notify from '../publishers/notify';
 import pushSw from '../publishers/push-sw';
-import queue from '../queue';
+import { createHttp } from '../queue';
 import watch from './watch';
 
 export default async (user, mentions, post) => {
@@ -84,7 +84,7 @@ export default async (user, mentions, post) => {
 				}
 
 				return new Promise((resolve, reject) => {
-					queue.create('http', {
+					createHttp({
 						type: 'deliverPost',
 						fromId: user._id,
 						toId: following.followerId,

src/queue.ts

-import { createQueue } from 'kue';
-import config from './config';
-
-export default createQueue({
-	redis: {
-		port: config.redis.port,
-		host: config.redis.host,
-		auth: config.redis.pass
-	}
-});

src/processor/index.ts → src/queue/index.ts

-import queue from '../queue';
-import db from './db';
-import http from './http';
+import { createQueue } from 'kue';
+import config from '../config';
+import db from './processors/db';
+import http from './processors/http';
 
-export default () => {
+const queue = createQueue({
+	redis: {
+		port: config.redis.port,
+		host: config.redis.host,
+		auth: config.redis.pass
+	}
+});
+
+export function createHttp(data) {
+	return queue
+		.create('http', data)
+		.attempts(16)
+		.backoff({ delay: 16384, type: 'exponential' });
+}
+
+export function createDb(data) {
+	return queue.create('db', data);
+}
+
+export function process() {
 	queue.process('db', db);
 
 	/*
@@ -15,4 +35,4 @@ export default () => {
 		http://kb.mozillazine.org/Network.http.max-connections
 	*/
 	queue.process('http', 256, http);
-};
+}

src/processor/db/delete-post-dependents.ts → src/queue/processors/db/delete-post-dependents.ts

-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 }),

src/processor/http/deliver-post.ts → src/queue/processors/http/deliver-post.ts

-import Post from '../../models/post';
-import User, { IRemoteUser } from '../../models/user';
-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 Post from '../../../models/post';
+import User, { IRemoteUser } from '../../../models/user';
+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 ({ data }) => {
 	const promisedTo = User.findOne({ _id: data.toId }) as Promise<IRemoteUser>;

src/processor/http/follow.ts → src/queue/processors/http/follow.ts

-import User, { isLocalUser, isRemoteUser, 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';
-import Logger from '../../utils/logger';
+import User, { isLocalUser, isRemoteUser, 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';
+import Logger from '../../../utils/logger';
 
 export default async ({ data }) => {
 	const { followerId, followeeId } = await Following.findOne({ _id: data.following });

src/processor/http/perform-activitypub.ts → src/queue/processors/http/perform-activitypub.ts

-import User from '../../models/user';
-import act from '../../remote/activitypub/act';
-import Resolver from '../../remote/activitypub/resolver';
+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))

src/processor/http/process-inbox.ts → src/queue/processors/http/process-inbox.ts

 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';
 
 export default async ({ data }): Promise<void> => {
 	const keyIdLower = data.signature.keyId.toLowerCase();

src/processor/http/report-github-failure.ts → src/queue/processors/http/report-github-failure.ts

 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';
+const createPost = require('../../../server/api/endpoints/posts/create');
 
 export default async ({ data }) => {
 	const asyncBot = User.findOne({ _id: data.userId });

src/processor/http/unfollow.ts → src/queue/processors/http/unfollow.ts

-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';
+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 });

src/remote/activitypub/act/follow.ts

@@ -3,7 +3,7 @@ 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 { createHttp } from '../../../queue';
 import context from '../renderer/context';
 import renderAccept from '../renderer/accept';
 import request from '../../request';
@@ -44,7 +44,7 @@ export default async (resolver: Resolver, actor, activity, distribute) => {
 		followerId: actor._id,
 		followeeId: followee._id
 	}).then(following => new Promise((resolve, reject) => {
-		queue.create('http', {
+		createHttp({
 			type: 'follow',
 			following: following._id
 		}).save(error => {

src/remote/activitypub/act/undo/unfollow.ts

-import queue from '../../../../queue';
+import { createHttp } from '../../../../queue';
 
 export default ({ $id }) => new Promise((resolve, reject) => {
-	queue.create('http', { type: 'unfollow', id: $id }).save(error => {
+	createHttp({ type: 'unfollow', id: $id }).save(error => {
 		if (error) {
 			reject(error);
 		} else {

src/remote/activitypub/delete/post.ts

 import Post from '../../../models/post';
-import queue from '../../../queue';
+import { createDb } from '../../../queue';
 
 export default async ({ $id }) => {
 	const promisedDeletion = Post.findOneAndDelete({ _id: $id });
 
-	await new Promise((resolve, reject) => queue.create('db', {
+	await new Promise((resolve, reject) => createDb({
 		type: 'deletePostDependents',
 		id: $id
 	}).delay(65536).save(error => error ? reject(error) : resolve()));

src/remote/activitypub/resolve-person.ts

 import { JSDOM } from 'jsdom';
 import { toUnicode } from 'punycode';
 import User, { validateUsername, isValidName, isValidDescription } from '../../models/user';
-import queue from '../../queue';
+import { createHttp } from '../../queue';
 import webFinger from '../webfinger';
 import create from './create';
 import Resolver from './resolver';
@@ -69,7 +69,7 @@ export default async (value, verifier?: string) => {
 		},
 	});
 
-	queue.create('http', {
+	createHttp({
 		type: 'performActivityPub',
 		actor: user._id,
 		outbox

src/server/activitypub/inbox.ts

 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',
 		inbox: req.body,
 		signature,

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 { createHttp } from '../../../../queue';
 
 /**
  * Follow a user
@@ -56,7 +56,7 @@ module.exports = (params, user) => new Promise(async (res, rej) => {
 		followeeId: followee._id
 	});
 
-	queue.create('http', { type: 'follow', following: _id }).save();
+	createHttp({ type: 'follow', following: _id }).save();
 
 	// Send response
 	res();