Skip to content
This repository was archived by the owner on Aug 12, 2023. It is now read-only.

Commit 68d6740

Browse files
cboviscbovis
authored
feat: start indexing app attributions after fill creation & measurement (#433)
* Create job constants for app attribution indexing * Add tradeCountContribution virtual to Fill model * Index app attributions after fill creation * Add additional tests for consumer * Index app attributions after fill measurement * Cover additional branch and fix issue
1 parent 13d0e71 commit 68d6740

10 files changed

Lines changed: 468 additions & 1 deletion

File tree

‎package-lock.json‎

Lines changed: 59 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎package.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
},
2626
"homepage": "https://0xtracker.com",
2727
"devDependencies": {
28+
"@elastic/elasticsearch-mock": "0.3.0",
2829
"eslint": "6.8.0",
2930
"eslint-config-airbnb-base": "14.0.0",
3031
"eslint-config-prettier": "6.9.0",

‎src/constants.js‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,7 @@ module.exports = {
5959
CREATE_TOKEN: 'create-token',
6060
FETCH_TOKEN_METADATA: 'fetch-token-metadata',
6161
FETCH_TRANSACTION: 'fetch-transaction',
62+
INDEX_APP_FILL_ATTRIBUTONS: 'index-app-fill-attributions',
6263
INDEX_FILL: 'index-fill',
6364
INDEX_FILL_PROTOCOL_FEE: 'index-fill-protocol-fee',
6465
INDEX_FILL_VALUE: 'index-fill-value',
@@ -67,6 +68,7 @@ module.exports = {
6768
QUEUE: {
6869
FILL_INDEXING: 'fill-indexing',
6970
FILL_PROCESSING: 'fill-processing',
71+
INDEXING: 'indexing',
7072
TOKEN_PROCESSING: 'token-processing',
7173
TRANSACTION_PROCESSING: 'transaction-processing',
7274
TRADED_TOKEN_INDEXING: 'traded-token-indexing',

‎src/consumers/index-app-fill-attributions.js‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,61 @@
1+
const _ = require('lodash');
2+
const signale = require('signale');
3+
4+
const { JOB, QUEUE } = require('../constants');
5+
const elasticsearch = require('../util/elasticsearch');
6+
7+
const logger = signale.scope('index app attributions for fill');
8+
9+
const consumer = async job => {
10+
const { attributions, date, fillId } = job.data;
11+
12+
logger.info(`indexing app attributions for fill: ${fillId}`);
13+
14+
const body = attributions
15+
.map(attribution => {
16+
return [
17+
JSON.stringify({
18+
update: {
19+
_id: `${attribution.appId}_${fillId}`,
20+
},
21+
}),
22+
JSON.stringify({
23+
doc: {
24+
appId: attribution.appId,
25+
date,
26+
fillId,
27+
relayedTrades: attribution.relayedTrades,
28+
relayedVolume: attribution.relayedVolume,
29+
sourcedTrades: attribution.sourcedTrades,
30+
sourcedVolume: attribution.sourcedVolume,
31+
totalTrades: attribution.totalTrades,
32+
totalVolume: attribution.totalVolume,
33+
updatedAt: new Date().toISOString(),
34+
},
35+
doc_as_upsert: true,
36+
}),
37+
].join('\n');
38+
})
39+
.join('\n');
40+
41+
const result = await elasticsearch
42+
.getClient()
43+
.bulk({ body: `${body}\n`, index: 'app_fill_attributions' });
44+
45+
if (result.body.errors === true) {
46+
const errorMessage = _.get(
47+
result,
48+
'body.items[0].update.error.reason',
49+
`Failed to index app attributions for fill: ${fillId}`,
50+
);
51+
throw new Error(errorMessage);
52+
}
53+
54+
logger.info(`indexed app attributions for fill: ${fillId}`);
55+
};
56+
57+
module.exports = {
58+
fn: consumer,
59+
jobName: JOB.INDEX_APP_FILL_ATTRIBUTONS,
60+
queueName: QUEUE.INDEXING,
61+
};

‎src/consumers/index-app-fill-attributions.test.js‎

Lines changed: 176 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,176 @@
1+
const { Client } = require('@elastic/elasticsearch');
2+
const Mock = require('@elastic/elasticsearch-mock');
3+
const timekeeper = require('timekeeper');
4+
5+
const elasticsearch = require('../util/elasticsearch');
6+
const consumer = require('./index-app-fill-attributions');
7+
8+
jest.mock('../util/elasticsearch');
9+
10+
const mock = new Mock();
11+
12+
beforeAll(() => {
13+
const client = new Client({
14+
node: 'http://localhost:9200',
15+
Connection: mock.getConnection(),
16+
});
17+
elasticsearch.getClient.mockReturnValue(client);
18+
});
19+
20+
afterEach(() => {
21+
mock.clearAll();
22+
});
23+
24+
describe('consumers/index-app-fill-attributions', () => {
25+
it('should consume indexing queue', () => {
26+
expect(consumer.queueName).toBe('indexing');
27+
});
28+
29+
it('should consume index-app-fill-attributions jobs', () => {
30+
expect(consumer.jobName).toBe('index-app-fill-attributions');
31+
});
32+
33+
it('should update documents for specified attributions', async () => {
34+
timekeeper.freeze('2020-08-02T08:42:24.934Z');
35+
mock.add(
36+
{
37+
method: 'POST',
38+
path: '/app_fill_attributions/_bulk',
39+
},
40+
({ body }) => {
41+
expect(body).toEqual([
42+
{
43+
update: {
44+
_id:
45+
'5980a15c-e450-40d3-8ef4-c54a37363ed0_5f267c7b545e125452c56e14',
46+
},
47+
},
48+
{
49+
doc: {
50+
appId: '5980a15c-e450-40d3-8ef4-c54a37363ed0',
51+
date: '2020-08-02T07:47:28.000Z',
52+
fillId: '5f267c7b545e125452c56e14',
53+
relayedTrades: 1,
54+
relayedVolume: 1520,
55+
totalTrades: 1,
56+
totalVolume: 1520,
57+
updatedAt: '2020-08-02T08:42:24.934Z',
58+
},
59+
doc_as_upsert: true,
60+
},
61+
{
62+
update: {
63+
_id:
64+
'449fcf3c-5f05-4cb4-b7bd-cf7c86bd6576_5f267c7b545e125452c56e14',
65+
},
66+
},
67+
{
68+
doc: {
69+
appId: '449fcf3c-5f05-4cb4-b7bd-cf7c86bd6576',
70+
date: '2020-08-02T07:47:28.000Z',
71+
fillId: '5f267c7b545e125452c56e14',
72+
sourcedTrades: 1,
73+
sourcedVolume: 1520,
74+
totalTrades: 1,
75+
totalVolume: 1520,
76+
updatedAt: '2020-08-02T08:42:24.934Z',
77+
},
78+
doc_as_upsert: true,
79+
},
80+
]);
81+
return { status: 'ok' };
82+
},
83+
);
84+
85+
await consumer.fn({
86+
data: {
87+
date: new Date('2020-08-02T07:47:28Z'),
88+
fillId: '5f267c7b545e125452c56e14',
89+
attributions: [
90+
{
91+
appId: '5980a15c-e450-40d3-8ef4-c54a37363ed0',
92+
relayedTrades: 1,
93+
relayedVolume: 1520,
94+
totalTrades: 1,
95+
totalVolume: 1520,
96+
},
97+
{
98+
appId: '449fcf3c-5f05-4cb4-b7bd-cf7c86bd6576',
99+
sourcedTrades: 1,
100+
sourcedVolume: 1520,
101+
totalTrades: 1,
102+
totalVolume: 1520,
103+
},
104+
],
105+
},
106+
});
107+
});
108+
109+
it('should update documents for specified attributions without volume', async () => {
110+
timekeeper.freeze('2020-08-02T08:42:24.934Z');
111+
mock.add(
112+
{
113+
method: 'POST',
114+
path: '/app_fill_attributions/_bulk',
115+
},
116+
({ body }) => {
117+
expect(body).toEqual([
118+
{
119+
update: {
120+
_id:
121+
'5980a15c-e450-40d3-8ef4-c54a37363ed0_5f267c7b545e125452c56e14',
122+
},
123+
},
124+
{
125+
doc: {
126+
appId: '5980a15c-e450-40d3-8ef4-c54a37363ed0',
127+
date: '2020-08-02T07:47:28.000Z',
128+
fillId: '5f267c7b545e125452c56e14',
129+
relayedTrades: 1,
130+
totalTrades: 1,
131+
updatedAt: '2020-08-02T08:42:24.934Z',
132+
},
133+
doc_as_upsert: true,
134+
},
135+
{
136+
update: {
137+
_id:
138+
'449fcf3c-5f05-4cb4-b7bd-cf7c86bd6576_5f267c7b545e125452c56e14',
139+
},
140+
},
141+
{
142+
doc: {
143+
appId: '449fcf3c-5f05-4cb4-b7bd-cf7c86bd6576',
144+
date: '2020-08-02T07:47:28.000Z',
145+
fillId: '5f267c7b545e125452c56e14',
146+
sourcedTrades: 1,
147+
totalTrades: 1,
148+
updatedAt: '2020-08-02T08:42:24.934Z',
149+
},
150+
doc_as_upsert: true,
151+
},
152+
]);
153+
return { status: 'ok' };
154+
},
155+
);
156+
157+
await consumer.fn({
158+
data: {
159+
date: new Date('2020-08-02T07:47:28Z'),
160+
fillId: '5f267c7b545e125452c56e14',
161+
attributions: [
162+
{
163+
appId: '5980a15c-e450-40d3-8ef4-c54a37363ed0',
164+
relayedTrades: 1,
165+
totalTrades: 1,
166+
},
167+
{
168+
appId: '449fcf3c-5f05-4cb4-b7bd-cf7c86bd6576',
169+
sourcedTrades: 1,
170+
totalTrades: 1,
171+
},
172+
],
173+
},
174+
});
175+
});
176+
});

0 commit comments

Comments
 (0)