event aggregate time periods api

.travis.yml

@@ -1,11 +1,21 @@
 services:
-        - mongodb
         - redis
 language: node_js
 node_js:
         - "10"
+before_script:
+        - mkdir -p ./database 
+        - mongod --fork --logpath /dev/null --dbpath ./database
+        - sleep 5
 script:
         - npm run lint
         - npm test
         - npm run test-integration
         #- npm run test-integration-external
+addons:
+  apt:
+    sources:
+        - mongodb-4.0-xenial
+    packages:
+        - mongodb-org-server
+        - mongodb-org-shell

routes/channel.js

@@ -19,6 +19,13 @@ router.get('/:id/validator-messages/:uid/:type?', channelIfExists, getValidatorM
 
 // event aggregates information
 router.get('/:id/events-aggregates', authRequired, channelLoad, getEventAggregates)
+// event aggregates with timeframe information
+router.get(
+	'/:id/events-aggregates/:earner',
+	celebrate({ body: schema.eventTimeAggr }),
+	channelLoad,
+	getEventTimeAggregate
+)
 
 // Submitting events/messages: requires auth
 router.post(
@@ -35,6 +42,61 @@ function getStatus(req, res) {
 	res.send({ channel: req.channel })
 }
 
+function getEventTimeAggregate(req, res, next) {
+	const { earner } = req.params
+	const {
+		eventType = 'IMPRESSION',
+		metric = 'eventCounts',
+		timeframe = 'year',
+		limit = 100
+	} = req.query
+	const eventsCol = db.getMongo().collection('eventAggregates')
+	const channel = req.channel
+	const group = getGroup(timeframe)
+
+	const pipeline = [
+		{
+			$addFields: {
+				value: {
+					$toInt: `$events.${eventType}.${metric}.${earner}`
+				}
+			}
+		},
+		{
+			$match: {
+				channelId: channel.id,
+				[`events.${eventType}.${metric}.${earner}`]: { $exists: true, $ne: null }
+			}
+		},
+		{
+			$project: {
+				value: 1,
+				created: 1,
+				year: { $year: '$created' },
+				month: { $month: '$created' },
+				week: { $week: '$created' },
+				day: { $dayOfMonth: '$created' },
+				hour: { $hour: '$created' },
+				minutes: { $minute: '$created' }
+			}
+		},
+		{ $sort: { created: 1 } },
+		{
+			$group: {
+				_id: { ...group },
+				value: { $sum: '$value' }
+			}
+		},
+		{ $limit: limit }
+	]
+
+	return eventsCol
+		.aggregate(pipeline)
+		.toArray()
+		.then(aggr => res.send({ channel, aggr }))
+		.catch(next)
+}
+
 function getEventAggregates(req, res, next) {
 	const eventsCol = db.getMongo().collection('eventAggregates')
 	const { uid } = req.session
@@ -207,5 +269,35 @@ function authRequired(req, res, next) {
 	next()
 }
 
+function getGroup(timeframe) {
+	if (timeframe === 'month') {
+		return { year: '$year', month: '$month' }
+	}
+
+	if (timeframe === 'week') {
+		return { year: '$year', week: '$week' }
+	}
+
+	if (timeframe === 'day') {
+		return { year: '$year', month: '$month', day: '$day' }
+	}
+
+	if (timeframe === 'hour') {
+		return { year: '$year', month: '$month', day: '$day', hour: '$hour' }
+	}
+
+	if (timeframe === 'minute') {
+		return {
+			year: '$year',
+			week: '$week',
+			month: '$month',
+			day: '$day',
+			hour: '$hour',
+			minutes: '$minutes'
+		}
+	}
+	return { year: '$year' }
+}
+
 // Export it
 module.exports = router

routes/schemas.js

@@ -148,5 +148,11 @@ module.exports = {
 					.required()
 			})
 		)
+	},
+	eventTimeAggr: {
+		eventType: Joi.string(),
+		metric: Joi.string().valid(['eventCounts', 'eventPayouts']),
+		timeframe: Joi.string().valid(['year', 'month', 'week', 'day', 'minute', 'hour']),
+		limit: Joi.number()
 	}
 }

test/integration.js

@@ -131,7 +131,7 @@ tape('new states are not produced when there are no new aggregates', async funct
 	t.end()
 })
 
-tape('/channel/{id}/events-aggregates', async function(t) {
+tape('/channel/{id}/events-aggregates/{earner}', async function(t) {
 	const id = 'eventAggregateCountTest'
 	const channel = { ...dummyVals.channel, id }
 
@@ -143,8 +143,8 @@ tape('/channel/{id}/events-aggregates', async function(t) {
 
 	// post events for that channel for multiple publishers
 	const publishers = [
-		[dummyVals.auth.publisher, genEvents(1, dummyVals.ids.publisher)],
-		[dummyVals.auth.publisher2, genEvents(1, dummyVals.ids.publisher2)]
+		[dummyVals.auth.creator, genEvents(3, dummyVals.ids.publisher)],
+		[dummyVals.auth.creator, genEvents(3, dummyVals.ids.publisher2)]
 	]
 
 	await Promise.all(
@@ -153,14 +153,14 @@ tape('/channel/{id}/events-aggregates', async function(t) {
 		)
 	)
 	await aggrAndTick()
-
 	const eventAggrFilterfixtures = [
 		// if we're a non superuser (validator) returns our event
 		[dummyVals.auth.publisher, 1],
 		[dummyVals.auth.publisher2, 1],
 		// if we're a superuser (validator) returns all events
 		[dummyVals.auth.leader, 2]
 	]
+
 	const url = `${leaderUrl}/channel/${id}/events-aggregates`
 
 	await Promise.all(
@@ -180,6 +180,32 @@ tape('/channel/{id}/events-aggregates', async function(t) {
 		})
 	)
 
+	const timeAggrFilterFixtures = [
+		['?metric=eventPayouts'],
+		['?metric=eventCounts'],
+		['?timeframe=minute'],
+		['?timeframe=year'],
+		['?timeframe=hour'],
+		['?timeframe=day'],
+		['?timeframe=week'],
+		['?timeframe=month']
+	]
+
+	await Promise.all(
+		timeAggrFilterFixtures.map(async fixture => {
+			const [query] = fixture
+			const resp = await fetch(`${url}/${dummyVals.ids.publisher}${query}`, {
+				method: 'GET',
+				headers: {
+					'content-type': 'application/json'
+				}
+			}).then(res => res.json())
+			t.ok(resp.channel, 'has resp.channel')
+			// 3 is number of events submitted by publisher in url param
+			t.ok(resp.aggr[0].value === 3, 'has correct aggr value')
+		})
+	)
+
 	t.end()
 })