Merge pull request #272 from samparsky/aip-6-rule

Aip 6 rule

routes/schemas.js

@@ -1,4 +1,5 @@
 const { Joi } = require('celebrate')
+const { eventTypes } = require('../services/constants')
 
 const numericString = Joi.string().regex(/^\d+$/)
 
@@ -70,6 +71,17 @@ const sentryValidatorMessage = Joi.object({
 	msg: Joi.array().items(validatorMessage)
 })
 
+const priceMultiplicationRules = Joi.array().items(
+	Joi.object({
+		multiplier: Joi.number().precision(10), // max 10 decimal places
+		amount: numericString,
+		evType: Joi.array().items(Joi.string().lowercase()),
+		country: Joi.array().items(Joi.string().lowercase()),
+		publisher: Joi.array().items(Joi.string()),
+		osType: Joi.array().items(Joi.string().lowercase())
+	})
+)
+
 module.exports = {
 	createChannel: {
 		id: Joi.string().required(),
@@ -120,16 +132,7 @@ module.exports = {
 			nonce: Joi.string(),
 			created: Joi.number(),
 			activeFrom: Joi.number(),
-			priceMultiplicationRules: Joi.array().items(
-				Joi.object({
-					multiplier: Joi.number().precision(10), // max 10 decimal places
-					amount: numericString,
-					evType: Joi.array().items(Joi.string().lowercase()),
-					country: Joi.array().items(Joi.string().lowercase()),
-					publisher: Joi.array().items(Joi.string()),
-					osType: Joi.array().items(Joi.string().lowercase())
-				})
-			),
+			priceMultiplicationRules,
 			priceDynamicAdjustment: Joi.bool()
 		}).required()
 	},
@@ -143,7 +146,11 @@ module.exports = {
 				publisher: Joi.string(),
 				ref: Joi.string().allow(''),
 				adUnit: Joi.string(),
-				adSlot: Joi.string()
+				adSlot: Joi.string(),
+				priceMultiplicationRules: priceMultiplicationRules.when('type', {
+					is: eventTypes.update_price_rules,
+					then: Joi.required()
+				})
 			})
 		)
 	},

services/constants.js

@@ -2,5 +2,8 @@ module.exports = Object.freeze({
 	collections: {
 		eventAggregates: 'eventAggregates',
 		analyticsAggregate: 'analyticsAggregate'
+	},
+	eventTypes: {
+		update_price_rules: 'PRICE_RULE_UPDATE'
 	}
 })

services/sentry/eventAggregator.js

@@ -5,6 +5,7 @@ const analyticsRecorder = require('./analyticsRecorder')
 const eventReducer = require('./lib/eventReducer')
 const checkAccess = require('./lib/access')
 const logger = require('../logger')('sentry')
+const { eventTypes } = require('../constants')
 
 const recorders = new Map()
 
@@ -21,7 +22,7 @@ function makeRecorder(channelId) {
 	const channelsCol = db.getMongo().collection('channels')
 
 	// get the channel
-	const channelPromise = channelsCol.findOne({ _id: channelId })
+	let channelPromise = channelsCol.findOne({ _id: channelId })
 
 	// persist each individual aggregate
 	// this is done in a one-at-a-time queue, with re-trying, to ensure everything is saved
@@ -55,6 +56,13 @@ function makeRecorder(channelId) {
 		trailing: true
 	})
 
+	const updateChannelPriceMultiplicationRules = async ev => {
+		await channelsCol.updateOne(
+			{ id: channelId },
+			{ $set: { 'spec.priceMultiplicationRules': ev.priceMultiplicationRules } }
+		)
+	}
+
 	// return a recorder
 	return async function(session, events) {
 		const channel = await channelPromise
@@ -64,6 +72,14 @@ function makeRecorder(channelId) {
 			return hasAccess
 		}
 
+		const priceRuleModifyEvs = events.filter(x => x.type === eventTypes.update_price_rules)
+		if (priceRuleModifyEvs.length) {
+			// if there are multiple evs only apply the latest
+			await updateChannelPriceMultiplicationRules(priceRuleModifyEvs[priceRuleModifyEvs.length - 1])
+
+			channelPromise = channelsCol.findOne({ _id: channel.id })
+		}
+
 		// No need to wait for this, it's simply a stats recorder
 		if (process.env.ANALYTICS_RECORDER) {
 			analyticsRecorder.record(channel, session, events)

services/sentry/lib/access.js

@@ -6,6 +6,7 @@ const cfg = require('../../../cfg')
 const redisCli = db.getRedis()
 const redisExists = promisify(redisCli.exists).bind(redisCli)
 const redisSetex = promisify(redisCli.setex).bind(redisCli)
+const { eventTypes } = require('../../constants')
 
 async function checkAccess(channel, session, events) {
 	const currentTime = Date.now()
@@ -23,8 +24,11 @@ async function checkAccess(channel, session, events) {
 	if (events.every(e => e.type === 'CLOSE') && (isCreator || isInWithdrawPeriod)) {
 		return { success: true }
 	}
-	// Only the creator can send a CLOSE
-	if (!isCreator && events.find(e => e.type === 'CLOSE')) {
+	// Only the creator can send a CLOSE & PRICE_RULE_UPDATE
+	if (
+		!isCreator &&
+		events.find(e => e.type === 'CLOSE' || e.type === eventTypes.update_price_rules)
+	) {
 		return { success: false, statusCode: 403 }
 	}
 

test/integration.js

@@ -17,6 +17,7 @@ const {
 } = require('./lib')
 const cfg = require('../cfg')
 const dummyVals = require('./prep-db/mongo')
+const constants = require('../services/constants')
 
 const postEvsAsCreator = (url, id, ev, headers = {}) =>
 	postEvents(url, id, ev, dummyVals.auth.creator, headers)
@@ -549,5 +550,45 @@ tape('analytics routes return correct values', async function(t) {
 	t.end()
 })
 
+tape('should update the priceMultiplicationRules', async function(t) {
+	const channel = getValidEthChannel()
+	channel.spec = {
+		...channel.spec,
+		pricingBounds: {
+			CLICK: {
+				min: '1',
+				max: '3'
+			}
+		},
+		priceMultiplicationRules: [{ amount: '2', country: ['US'], evType: ['CLICK'] }]
+	}
+	const num = 66
+	const evs = genEvents(num, randomAddress(), 'CLICK')
+	// Submit a new channel; we submit it to both sentries to avoid 404 when propagating messages
+	await Promise.all([
+		fetchPost(`${leaderUrl}/channel`, dummyVals.auth.leader, channel),
+		fetchPost(`${followerUrl}/channel`, dummyVals.auth.follower, channel)
+	])
+
+	// update priceMultiplicationRule
+	await fetchPost(`${leaderUrl}/channel/${channel.id}/events`, dummyVals.auth.creator, {
+		events: [
+			{
+				type: constants.eventTypes.update_price_rules,
+				priceMultiplicationRules: [{ amount: '3', country: ['US'], evType: ['CLICK'] }]
+			}
+		]
+	})
+
+	await postEvsAsCreator(leaderUrl, channel.id, evs, { 'cf-ipcountry': 'US' })
+	// Technically we don't need to tick, since the events should be reflected immediately
+	const analytics = await fetch(
+		`${leaderUrl}/analytics/${channel.id}?eventType=CLICK&metric=eventPayouts`
+	).then(r => r.json())
+	t.equal(analytics.aggr[0].value, (num * 3).toString(), 'proper payout amount')
+
+	t.end()
+})
+
 // @TODO sentry tests: ensure every middleware case is accounted for: channelIfExists, channelIfActive, auth
 // @TODO tests for the adapters and especially ewt

test/routes.js

@@ -12,6 +12,7 @@ const {
 	getValidEthChannel,
 	randomAddress
 } = require('./lib')
+const { eventTypes } = require('../services/constants')
 
 // const cfg = require('../cfg')
 const dummyVals = require('./prep-db/mongo')
@@ -132,6 +133,22 @@ tape('POST /channel/{id}/events: CLOSE: a publisher but not a creator', async fu
 	t.end()
 })
 
+tape(
+	`POST /channel/{id}/events: ${eventTypes.update_price_rules}: a publisher but not a creator`,
+	async function(t) {
+		await fetchPost(
+			`${leaderUrl}/channel/${dummyVals.channel.id}/events`,
+			dummyVals.auth.publisher,
+			{
+				events: [{ type: eventTypes.update_price_rules, priceMultiplicationRules: [] }]
+			}
+		).then(function(resp) {
+			t.equal(resp.status, 403, 'status must be Forbidden')
+		})
+		t.end()
+	}
+)
+
 tape('POST /channel/validate: invalid schema', async function(t) {
 	const resp = await fetchPost(`${followerUrl}/channel/validate`, dummyVals.auth.leader, {}).then(
 		r => r.json()