import '../loadenv';

import stream from 'getstream';
import moment from 'moment';
import normalize from 'normalize-url';

import RSS from '../models/rss';
import Article from '../models/article';

import '../utils/db';
import config from '../config';
import logger from '../utils/logger';

import sendRssFeedToCollections from '../utils/events/sendRssFeedToCollections';
import { ParseFeed } from '../parsers/feed';

import { ProcessRssQueue, OgQueueAdd } from '../asyncTasks';
import { getStatsDClient, timeIt } from '../utils/statsd';
import { upsertManyPosts } from '../utils/upsert';

const streamClient = stream.connect(config.stream.apiKey, config.stream.apiSecret);
const duplicateKeyError = 11000;

// connect the handler to the queue
logger.info('Starting the RSS worker');

//TODO: move this to a separate main.js
ProcessRssQueue(100, rssProcessor);

const statsd = getStatsDClient();

export async function rssProcessor(job) {
	logger.info(`Processing ${job.data.url}`);
	// just intercept error handling before it goes to Bull
	try {
		await handleRSS(job);
	} catch (err) {
		let tags = { queue: 'rss' };
		let extra = {
			JobRSS: job.data.rss,
			JobURL: job.data.url,
		};
		logger.error('RSS job encountered an error', { err, tags, extra });
	}
	logger.info(`Completed scraping for ${job.data.url}`);
}

// Handle Podcast scrapes the podcast and updates the episodes
export async function handleRSS(job) {
	let rssID = job.data.rss;

	await timeIt('winds.handle_rss.ack', () => {
		return markDone(rssID);
	});

	let rss = await timeIt('winds.handle_rss.get_rss', () => {
		return RSS.findOne({ _id: rssID });
	});

	if (!rss) {
		logger.warn(`RSS with ID ${rssID} does not exist`);
		return;
	}

	logger.info(`Marked ${rssID} as done`);

	// parse the articles
	let rssContent = await timeIt('winds.handle_rss.parsing', async () => {
		try {
			const res = await ParseFeed(job.data.url);
			await RSS.resetScrapeFailures(rssID);
			return res;
		} catch (err) {
			await RSS.incrScrapeFailures(rssID);
			throw new Error(`http request failed for url ${job.data.url}`);
		}
	});

	if (!rssContent) {
		return;
	}

	// update the articles
	logger.info(`Updating ${rssContent.articles.length} articles for feed ${rssID}`);

	if (rssContent.articles.length === 0) {
		return;
	}

	statsd.increment('winds.handle_rss.articles.parsed', rssContent.articles.length);
	statsd.timing('winds.handle_rss.articles.parsed', rssContent.articles.length);

	let updatedArticles
	let allArticles = await timeIt('winds.handle_rss.upsertManyArticles', async () => {
		let articles = rssContent.articles;
		for (let a of articles) {
			a.rss = rssID
		}
		let operationMap = await upsertManyPosts(rssID, articles, 'rss')
		updatedArticles = operationMap.new.concat(operationMap.changed)
		logger.info(`Finished updating. ${updatedArticles.length} out of ${articles.length} changed`)
	});

	// update the count
	await RSS.update(
		{ _id: rssID },
		{
			postCount: await Article.count({rss: rssID}),
			fingerprint: rssContent.fingerprint,
		}
	);

	statsd.increment('winds.handle_rss.articles.upserted', updatedArticles.length);

	await timeIt('winds.handle_rss.OgQueueAdd', () => {
		return Promise.all(
			updatedArticles.map(article => {
				OgQueueAdd(
					{
						type: 'article',
						url: article.url,
					},
					{
						removeOnComplete: true,
						removeOnFail: true,
					},
				);
			}),
		);
	});

	let t0 = new Date();
	let rssFeed = streamClient.feed('rss', rssID);
	logger.info(`Syncing ${updatedArticles.length} articles to Stream`);
	if (updatedArticles.length > 0) {
		let chunkSize = 100;
		for (let i = 0, j = updatedArticles.length; i < j; i += chunkSize) {
			let chunk = updatedArticles.slice(i, i + chunkSize);
			let streamArticles = chunk.map(article => {
				return {
					actor: article.rss,
					foreign_id: `articles:${article._id}`,
					object: article._id,
					time: article.publicationDate,
					verb: 'rss_article',
				};
			});
			await rssFeed.addActivities(streamArticles);
		}
		await sendRssFeedToCollections(rss);
	}
	statsd.timing('winds.handle_rss.send_to_stream', new Date() - t0);
}


// markDone sets lastScraped to now and isParsing to false
async function markDone(rssID) {
	const now = moment().toISOString();
	return await RSS.update(
		{ _id: rssID },
		{ lastScraped: now, isParsing: false },
	);
}

@@ -2456,115 +2456,8 @@
 );%0A%0A
-%09let updatedArticles%0A%09let allArticles = await timeIt('winds.handle_rss.upsertManyArticles', async () =%3E %7B%0A%09
 %09let
@@ -2487,19 +2487,17 @@
 articles
-;
 %0A
-%09
 %09for (le
@@ -2517,17 +2517,16 @@
 es) %7B%0A%09%09
-%09
 a.rss = 
@@ -2531,21 +2531,19 @@
 = rssID%0A
-%09
 %09%7D%0A
-%09
 %09let ope
@@ -2601,17 +2601,20 @@
 'rss')%0A%09
-%09
+let 
 updatedA
@@ -2669,17 +2669,16 @@
 hanged)%0A
-%09
 %09logger.
@@ -2763,21 +2763,16 @@
 hanged%60)
-%0A%09%7D);
 %0A%0A%09// up

remove timer to reduce things that could cause this error