Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Some advancments with ES queries #473

Merged
merged 8 commits into from
Jul 20, 2023
Merged

Some advancments with ES queries #473

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
4514046
Started with searching from ES
andris9 Jun 22, 2023
8c38e2e
fixed wrong path
andris9 Jun 22, 2023
2cb80ca
somewhat working ES query
andris9 Jul 17, 2023
744068d
Allow including specific header fields in message listing
andris9 Jul 20, 2023
8acdc03
fixed tests
andris9 Jul 20, 2023
c96174e
fixed tests
andris9 Jul 20, 2023
f711ec7
Bumped deps
andris9 Jul 20, 2023
9ac5b61
Bumped deps
andris9 Jul 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions docs/api/openapi.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1711,6 +1711,12 @@ paths:
description: Ordering of the records by insert date
schema:
$ref: '#/components/schemas/Order'
- name: includeHeaders
in: query
description: 'Comma separated list of header keys to include in the response'
schema:
type: string
example: 'List-ID, MIME-version'
- name: next
in: query
description: 'Cursor value for next page, retrieved from nextCursor response value'
Expand Down Expand Up @@ -2037,6 +2043,12 @@ paths:
description: Ordering of the records by insert date. If no order is supplied, results are sorted by heir mongoDB ObjectId.
schema:
$ref: '#/components/schemas/Order'
- name: includeHeaders
in: query
description: 'Comma separated list of header keys to include in the response'
schema:
type: string
example: 'List-ID, MIME-version'
- name: page
in: query
description: 'Current page number. Informational only, page numbers start from 1'
Expand Down Expand Up @@ -6918,6 +6930,9 @@ components:
metaData:
type: object
description: Custom metadata value. Included if metaData query argument was true
headers:
type: object
description: Header object keys requested with the includeHeaders argument
GetFilesResult:
required:
- id
Expand Down
13 changes: 11 additions & 2 deletions indexer.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ class Indexer {
this.running = false;
log.info('Indexer', 'Stopping indexer');
try {
if (this.changeStream && !this.changeStream.isClosed()) {
if (this.changeStream && !this.changeStream.closed) {
await this.changeStream.close();
}
} catch (err) {
Expand Down Expand Up @@ -188,7 +188,14 @@ class Indexer {
return;
}

if (this.changeStream.isClosed()) {
if (error.errorLabels && error.errorLabels.includes('NonResumableChangeStreamError')) {
// can't resume previous cursor
await db.redis.del('indexer:last');
log.info('Indexer', 'Can not resume existing cursor');
return;
}

if (this.changeStream && this.changeStream.closed) {
log.info('Indexer', 'The change stream is closed. Will not wait on any more changes.');
return;
} else {
Expand Down Expand Up @@ -335,6 +342,8 @@ function indexingJob(esclient) {
uid: messageData.uid,
answered: messageData.flags ? messageData.flags.includes('\\Answered') : null,

ha: (messageData.attachments && messageData.attachments.length > 0) || false,

attachments:
(messageData.attachments &&
messageData.attachments.map(attachment =>
Expand Down
152 changes: 126 additions & 26 deletions lib/api/messages.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,8 @@ const { nextPageCursorSchema, previousPageCursorSchema, pageNrSchema, sessSchema
const { preprocessAttachments } = require('../data-url');
const TaskHandler = require('../task-handler');
const prepareSearchFilter = require('../prepare-search-filter');
const { getMongoDBQuery } = require('../search-query');
const { getMongoDBQuery /*, getElasticSearchQuery*/ } = require('../search-query');
//const { getClient } = require('../elasticsearch');

const BimiHandler = require('../bimi-handler');

Expand Down Expand Up @@ -386,6 +387,8 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
let sortAscending = result.value.order === 'asc';
let filterUnseen = result.value.unseen;

let includeHeaders = result.value.includeHeaders ? result.value.includeHeaders.split(',') : false;

let mailboxData;
try {
mailboxData = await db.database.collection('mailboxes').findOne(
Expand Down Expand Up @@ -442,13 +445,6 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
hdate: true,
idate: true,
subject: true,
'mimeTree.parsedHeader.from': true,
'mimeTree.parsedHeader.sender': true,
'mimeTree.parsedHeader.to': true,
'mimeTree.parsedHeader.cc': true,
'mimeTree.parsedHeader.bcc': true,
'mimeTree.parsedHeader.content-type': true,
'mimeTree.parsedHeader.references': true,
ha: true,
size: true,
intro: true,
Expand All @@ -465,6 +461,24 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
sortAscending
};

if (includeHeaders) {
// get all headers
opts.fields.projection['mimeTree.parsedHeader'] = true;
} else {
// get only required headers
for (let requiredHeader of [
'mimeTree.parsedHeader.from',
'mimeTree.parsedHeader.sender',
'mimeTree.parsedHeader.to',
'mimeTree.parsedHeader.cc',
'mimeTree.parsedHeader.bcc',
'mimeTree.parsedHeader.content-type',
'mimeTree.parsedHeader.references'
]) {
opts.fields.projection[requiredHeader] = true;
}
}

if (pageNext) {
opts.next = pageNext;
} else if ((!page || page > 1) && pagePrevious) {
Expand Down Expand Up @@ -499,7 +513,7 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
previousCursor: listing.hasPrevious ? listing.previous : false,
nextCursor: listing.hasNext ? listing.next : false,
specialUse: mailboxData.specialUse,
results: (listing.results || []).map(formatMessageListing)
results: (listing.results || []).map(entry => formatMessageListing(entry, includeHeaders))
};

return res.json(response);
Expand Down Expand Up @@ -531,6 +545,12 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
attachments: booleanSchema,
flagged: booleanSchema,
unseen: booleanSchema,
includeHeaders: Joi.string()
.max(1024)
.trim()
.empty('')
.example('List-ID, MIME-Version')
.description('Comma separated list of header keys to include in the response'),
searchable: booleanSchema,
sess: sessSchema,
ip: sessIPSchema
Expand All @@ -545,6 +565,12 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
threadCounters: booleanSchema.default(false),
limit: Joi.number().default(20).min(1).max(250),
order: Joi.any().empty('').allow('asc', 'desc').optional(),
includeHeaders: Joi.string()
.max(1024)
.trim()
.empty('')
.example('List-ID, MIME-Version')
.description('Comma separated list of header keys to include in the response'),
next: nextPageCursorSchema,
previous: previousPageCursorSchema,
page: pageNrSchema
Expand Down Expand Up @@ -580,10 +606,35 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
let pagePrevious = result.value.previous;
let order = result.value.order;

let includeHeaders = result.value.includeHeaders ? result.value.includeHeaders.split(',') : false;

let filter;
let query;

if (result.value.q) {
let hasESFeatureFlag = await db.redis.sismember(`feature:indexing`, user.toString());
if (hasESFeatureFlag) {
// search from ElasticSearch
/*
// TODO: paging and cursors for ElasticSearch results

let searchQuery = await getElasticSearchQuery(db, user, result.value.q);

const esclient = getClient();

const searchOpts = {
index: config.elasticsearch.index,
body: { query: searchQuery, sort: { uid: 'desc' } }
};

let searchResult = await esclient.search(searchOpts);
const searchHits = searchResult && searchResult.body && searchResult.body.hits;

console.log('ES RESULTS');
console.log(util.inspect(searchResult, false, 22, true));
*/
}

filter = await getMongoDBQuery(db, user, result.value.q);
query = result.value.q;
} else {
Expand Down Expand Up @@ -611,13 +662,6 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
hdate: true,
idate: true,
subject: true,
'mimeTree.parsedHeader.from': true,
'mimeTree.parsedHeader.sender': true,
'mimeTree.parsedHeader.to': true,
'mimeTree.parsedHeader.cc': true,
'mimeTree.parsedHeader.bcc': true,
'mimeTree.parsedHeader.content-type': true,
'mimeTree.parsedHeader.references': true,
ha: true,
intro: true,
size: true,
Expand All @@ -634,6 +678,24 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
sortAscending: order === 'asc' ? true : undefined
};

if (includeHeaders) {
// get all headers
opts.fields.projection['mimeTree.parsedHeader'] = true;
} else {
// get only required headers
for (let requiredHeader of [
'mimeTree.parsedHeader.from',
'mimeTree.parsedHeader.sender',
'mimeTree.parsedHeader.to',
'mimeTree.parsedHeader.cc',
'mimeTree.parsedHeader.bcc',
'mimeTree.parsedHeader.content-type',
'mimeTree.parsedHeader.references'
]) {
opts.fields.projection[requiredHeader] = true;
}
}

if (pageNext) {
opts.next = pageNext;
} else if ((!page || page > 1) && pagePrevious) {
Expand Down Expand Up @@ -668,7 +730,7 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
page,
previousCursor: listing.hasPrevious ? listing.previous : false,
nextCursor: listing.hasNext ? listing.next : false,
results: (listing.results || []).map(formatMessageListing)
results: (listing.results || []).map(entry => formatMessageListing(entry, includeHeaders))
};

return res.json(response);
Expand Down Expand Up @@ -2470,6 +2532,12 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
next: nextPageCursorSchema,
previous: previousPageCursorSchema,
order: Joi.any().empty('').allow('asc', 'desc').default('desc'),
includeHeaders: Joi.string()
.max(1024)
.trim()
.empty('')
.example('List-ID, MIME-Version')
.description('Comma separated list of header keys to include in the response'),
page: pageNrSchema,
sess: sessSchema,
ip: sessIPSchema
Expand Down Expand Up @@ -2504,6 +2572,8 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
let pagePrevious = result.value.previous;
let sortAscending = result.value.order === 'asc';

let includeHeaders = result.value.includeHeaders ? result.value.includeHeaders.split(',') : false;

let total = await db.database.collection('archived').countDocuments({ user });

let opts = {
Expand All @@ -2522,13 +2592,6 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
hdate: true,
idate: true,
subject: true,
'mimeTree.parsedHeader.from': true,
'mimeTree.parsedHeader.sender': true,
'mimeTree.parsedHeader.to': true,
'mimeTree.parsedHeader.cc': true,
'mimeTree.parsedHeader.bcc': true,
'mimeTree.parsedHeader.content-type': true,
'mimeTree.parsedHeader.references': true,
ha: true,
intro: true,
size: true,
Expand All @@ -2544,6 +2607,24 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
sortAscending
};

if (includeHeaders) {
// get all headers
opts.fields.projection['mimeTree.parsedHeader'] = true;
} else {
// get only required headers
for (let requiredHeader of [
'mimeTree.parsedHeader.from',
'mimeTree.parsedHeader.sender',
'mimeTree.parsedHeader.to',
'mimeTree.parsedHeader.cc',
'mimeTree.parsedHeader.bcc',
'mimeTree.parsedHeader.content-type',
'mimeTree.parsedHeader.references'
]) {
opts.fields.projection[requiredHeader] = true;
}
}

if (pageNext) {
opts.next = pageNext;
} else if ((!page || page > 1) && pagePrevious) {
Expand Down Expand Up @@ -2577,7 +2658,7 @@ module.exports = (db, server, messageHandler, userHandler, storageHandler, setti
m.uid = m._id;
return m;
})
.map(formatMessageListing)
.map(entry => formatMessageListing(entry, includeHeaders))
};

return res.json(response);
Expand Down Expand Up @@ -3098,7 +3179,17 @@ function leftPad(val, chr, len) {
return chr.repeat(len - val.toString().length) + val;
}

function formatMessageListing(messageData) {
function formatMessageListing(messageData, includeHeaders) {
includeHeaders = []
.concat(includeHeaders || [])
.map(entry => {
if (typeof entry !== 'string') {
return false;
}
return entry.toLowerCase().trim();
})
.filter(entry => entry);

let parsedHeader = (messageData.mimeTree && messageData.mimeTree.parsedHeader) || {};

let from = parsedHeader.from ||
Expand Down Expand Up @@ -3147,6 +3238,15 @@ function formatMessageListing(messageData) {
bimi: messageData.bimi
};

if (includeHeaders.length) {
response.headers = {};
for (let headerKey of includeHeaders) {
if (parsedHeader[headerKey]) {
response.headers[headerKey] = parsedHeader[headerKey];
}
}
}

if (messageData.meta && 'custom' in messageData.meta) {
response.metaData = tools.formatMetaData(messageData.meta.custom);
}
Expand Down
2 changes: 2 additions & 0 deletions lib/api/users.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -859,6 +859,8 @@ module.exports = (db, server, userHandler, settingsHandler) => {

fromWhitelist: userData.fromWhitelist || [],

featureFlags: userData.featureFlags || {},

disabledScopes: userData.disabledScopes || [],

hasPasswordSet: !!userData.password || !!userData.tempPassword,
Expand Down
5 changes: 5 additions & 0 deletions lib/ensure-es-index.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,11 @@ const mappings = {
type: 'boolean'
},

// has attachments
ha: {
type: 'boolean'
},

attachments: {
type: 'nested',
properties: {
Expand Down
Loading
Loading