feat(ElasticSearch): Postgres synchronization

config/default.json

@@ -1,5 +1,9 @@
 {
   "port": "3060",
+  "services": {
+    "server": true,
+    "searchSync": true
+  },
   "mailpit": {
     "client": false
   },

config/test.json

@@ -1,5 +1,9 @@
 {
   "port": "3061",
+  "services": {
+    "server": true,
+    "searchSync": false
+  },
   "database": {
     "url": "postgres://opencollective@127.0.0.1:5432/opencollective_test"
   },

package.json

@@ -116,6 +116,7 @@
     "pg": "8.13.1",
     "pg-connection-string": "2.7.0",
     "pg-format": "1.0.4",
+    "pg-listen": "1.7.0",
     "plaid": "29.0.0",
     "prepend-http": "3.0.1",
     "redis": "4.6.6",

server/index.js

@@ -7,14 +7,16 @@ import config from 'config';
 import express from 'express';
 import throng from 'throng';
 
+import { startElasticSearchPostgresSync } from './lib/elastic-search/sync-postgres';
 import expressLib from './lib/express';
 import logger from './lib/logger';
 import { updateCachedFidoMetadata } from './lib/two-factor-authentication/fido-metadata';
+import { parseToBoolean } from './lib/utils';
 import routes from './routes';
 
 const workers = process.env.WEB_CONCURRENCY || 1;
 
-async function start(i) {
+async function startExpressServer(workerId) {
   const expressApp = express();
 
   await updateCachedFidoMetadata();
@@ -35,7 +37,7 @@ async function start(i) {
       host,
       server.address().port,
       config.env,
-      i,
+      workerId,
     );
   });
 
@@ -45,15 +47,22 @@ async function start(i) {
   return expressApp;
 }
 
+// Start the express server
 let app;
+if (parseToBoolean(config.services.server)) {
+  if (['production', 'staging'].includes(config.env) && workers > 1) {
+    throng({ worker: startExpressServer, count: workers });
+  } else {
+    app = startExpressServer(1);
+  }
+}
 
-if (['production', 'staging'].includes(config.env) && workers > 1) {
-  throng({ worker: start, count: workers });
-} else {
-  app = start(1);
+// Start the search sync job
+if (parseToBoolean(config.services.searchSync)) {
+  startElasticSearchPostgresSync();
 }
 
 // This is used by tests
-export default async function () {
-  return app ? app : start(1);
+export default async function startServerForTest() {
+  return app ? app : parseToBoolean(config.services.server) ? startExpressServer(1) : null;
 }

server/lib/elastic-search/adapters/ElasticSearchCollectivesAdapter.ts

@@ -1,11 +1,12 @@
-import models, { Op } from '../../../models';
+import { Op } from '../../../models';
+import Collective from '../../../models/Collective';
 import { stripHTMLOrEmpty } from '../../sanitize-html';
 import { ElasticSearchIndexName } from '../constants';
 
 import { ElasticSearchModelAdapter } from './ElasticSearchModelAdapter';
 
 export class ElasticSearchCollectivesAdapter implements ElasticSearchModelAdapter {
-  public readonly model = models.Collective;
+  public readonly model = Collective;
   public readonly index = ElasticSearchIndexName.COLLECTIVES;
   public readonly mappings = {
     properties: {
@@ -38,8 +39,8 @@ export class ElasticSearchCollectivesAdapter implements ElasticSearchModelAdapte
       maxId?: number;
       ids?: number[];
     } = {},
-  ): Promise<Array<InstanceType<typeof models.Collective>>> {
-    return models.Collective.findAll({
+  ): Promise<Array<InstanceType<typeof Collective>>> {
+    return Collective.findAll({
       attributes: Object.keys(this.mappings.properties),
       order: [['id', 'DESC']],
       limit: options.limit,
@@ -54,7 +55,7 @@ export class ElasticSearchCollectivesAdapter implements ElasticSearchModelAdapte
   }
 
   public mapModelInstanceToDocument(
-    instance: InstanceType<typeof models.Collective>,
+    instance: InstanceType<typeof Collective>,
   ): Record<keyof (typeof this.mappings)['properties'], unknown> {
     return {
       id: instance.id,
@@ -72,7 +73,7 @@ export class ElasticSearchCollectivesAdapter implements ElasticSearchModelAdapte
       isActive: instance.isActive,
       isHostAccount: instance.isHostAccount,
       deactivatedAt: instance.deactivatedAt,
-      HostCollectiveId: instance.HostCollectiveId,
+      HostCollectiveId: !instance.isActive ? null : instance.HostCollectiveId,
       ParentCollectiveId: instance.ParentCollectiveId,
     };
   }

server/lib/elastic-search/adapters/ElasticSearchCommentsAdapter.ts

@@ -2,15 +2,14 @@ import { QueryDslQueryContainer } from '@elastic/elasticsearch/lib/api/types';
 import { omit } from 'lodash';
 import { Op } from 'sequelize';
 
-import models from '../../../models';
-import { CommentType } from '../../../models/Comment';
+import Comment, { CommentType } from '../../../models/Comment';
 import { stripHTMLOrEmpty } from '../../sanitize-html';
 import { ElasticSearchIndexName } from '../constants';
 
 import { ElasticSearchModelAdapter } from './ElasticSearchModelAdapter';
 
 export class ElasticSearchCommentsAdapter implements ElasticSearchModelAdapter {
-  public readonly model = models.Comment;
+  public readonly model = Comment;
   public readonly index = ElasticSearchIndexName.COMMENTS;
   public readonly mappings = {
     properties: {
@@ -38,7 +37,7 @@ export class ElasticSearchCommentsAdapter implements ElasticSearchModelAdapter {
       ids?: number[];
     } = {},
   ) {
-    return models.Comment.findAll({
+    return Comment.findAll({
       attributes: omit(Object.keys(this.mappings.properties), ['HostCollectiveId', 'ParentCollectiveId']),
       order: [['id', 'DESC']],
       limit: options.limit,
@@ -52,7 +51,7 @@ export class ElasticSearchCommentsAdapter implements ElasticSearchModelAdapter {
         {
           association: 'collective',
           required: true,
-          attributes: ['HostCollectiveId', 'ParentCollectiveId'],
+          attributes: ['isActive', 'HostCollectiveId', 'ParentCollectiveId'],
         },
         {
           association: 'expense',
@@ -69,7 +68,7 @@ export class ElasticSearchCommentsAdapter implements ElasticSearchModelAdapter {
   }
 
   public mapModelInstanceToDocument(
-    instance: InstanceType<typeof models.Comment>,
+    instance: InstanceType<typeof Comment>,
   ): Record<keyof (typeof this.mappings)['properties'], unknown> {
     return {
       id: instance.id,
@@ -84,7 +83,7 @@ export class ElasticSearchCommentsAdapter implements ElasticSearchModelAdapter {
       HostCollectiveId:
         instance.expense?.HostCollectiveId ??
         instance.hostApplication?.HostCollectiveId ??
-        instance.collective.HostCollectiveId,
+        (!instance.collective.isActive ? null : instance.collective.HostCollectiveId),
     };
   }
 

server/lib/elastic-search/adapters/ElasticSearchExpensesAdapter.ts

@@ -1,14 +1,14 @@
 import { omit } from 'lodash';
 import { Op } from 'sequelize';
 
-import models from '../../../models';
+import Expense from '../../../models/Expense';
 import { stripHTMLOrEmpty } from '../../sanitize-html';
 import { ElasticSearchIndexName } from '../constants';
 
 import { ElasticSearchModelAdapter } from './ElasticSearchModelAdapter';
 
 export class ElasticSearchExpensesAdapter implements ElasticSearchModelAdapter {
-  public readonly model = models.Expense;
+  public readonly model = Expense;
   public readonly index = ElasticSearchIndexName.EXPENSES;
   public readonly mappings = {
     properties: {
@@ -23,6 +23,7 @@ export class ElasticSearchExpensesAdapter implements ElasticSearchModelAdapter {
       amount: { type: 'integer' },
       currency: { type: 'keyword' },
       status: { type: 'keyword' },
+      items: { type: 'text' },
       // Relationships
       UserId: { type: 'keyword' },
       CollectiveId: { type: 'keyword' },
@@ -42,7 +43,7 @@ export class ElasticSearchExpensesAdapter implements ElasticSearchModelAdapter {
       ids?: number[];
     } = {},
   ) {
-    return models.Expense.findAll({
+    return Expense.findAll({
       attributes: omit(Object.keys(this.mappings.properties), ['ParentCollectiveId']),
       order: [['id', 'DESC']],
       limit: options.limit,
@@ -56,14 +57,19 @@ export class ElasticSearchExpensesAdapter implements ElasticSearchModelAdapter {
         {
           association: 'collective',
           required: true,
-          attributes: ['HostCollectiveId', 'ParentCollectiveId'],
+          attributes: ['isActive', 'HostCollectiveId', 'ParentCollectiveId'],
+        },
+        {
+          association: 'items',
+          required: true,
+          attributes: ['description'],
         },
       ],
     });
   }
 
   public mapModelInstanceToDocument(
-    instance: InstanceType<typeof models.Expense>,
+    instance: InstanceType<typeof Expense>,
   ): Record<keyof (typeof this.mappings)['properties'], unknown> {
     return {
       id: instance.id,
@@ -74,14 +80,16 @@ export class ElasticSearchExpensesAdapter implements ElasticSearchModelAdapter {
       privateMessage: stripHTMLOrEmpty(instance.privateMessage),
       invoiceInfo: instance.invoiceInfo,
       reference: instance.reference,
+      items: instance.items.map(item => item.description).join(' '),
       amount: instance.amount,
       currency: instance.currency,
       status: instance.status,
       CollectiveId: instance.CollectiveId,
       ParentCollectiveId: instance.collective.ParentCollectiveId,
       FromCollectiveId: instance.FromCollectiveId,
       UserId: instance.UserId,
-      HostCollectiveId: instance.HostCollectiveId || instance.collective.HostCollectiveId,
+      HostCollectiveId:
+        instance.HostCollectiveId || (!instance.collective.isActive ? null : instance.collective.HostCollectiveId),
     };
   }
 

server/lib/elastic-search/adapters/ElasticSearchHostApplicationsAdapter.ts

@@ -1,13 +1,13 @@
 import { omit } from 'lodash';
 import { Op } from 'sequelize';
 
-import models from '../../../models';
+import HostApplication from '../../../models/HostApplication';
 import { ElasticSearchIndexName } from '../constants';
 
 import { ElasticSearchModelAdapter } from './ElasticSearchModelAdapter';
 
 export class ElasticSearchHostApplicationsAdapter implements ElasticSearchModelAdapter {
-  public readonly model = models.HostApplication;
+  public readonly model = HostApplication;
   public readonly index = ElasticSearchIndexName.HOST_APPLICATIONS;
   public readonly mappings = {
     properties: {
@@ -33,7 +33,7 @@ export class ElasticSearchHostApplicationsAdapter implements ElasticSearchModelA
       ids?: number[];
     } = {},
   ) {
-    return models.HostApplication.findAll({
+    return HostApplication.findAll({
       attributes: omit(Object.keys(this.mappings.properties), ['ParentCollectiveId']),
       order: [['id', 'DESC']],
       limit: options.limit,
@@ -54,7 +54,7 @@ export class ElasticSearchHostApplicationsAdapter implements ElasticSearchModelA
   }
 
   public mapModelInstanceToDocument(
-    instance: InstanceType<typeof models.HostApplication>,
+    instance: InstanceType<typeof HostApplication>,
   ): Record<keyof (typeof this.mappings)['properties'], unknown> {
     return {
       id: instance.id,

server/lib/elastic-search/adapters/ElasticSearchOrdersAdapter.ts

@@ -1,13 +1,13 @@
 import { omit } from 'lodash';
 import { Op } from 'sequelize';
 
-import models from '../../../models';
+import Order from '../../../models/Order';
 import { ElasticSearchIndexName } from '../constants';
 
 import { ElasticSearchModelAdapter } from './ElasticSearchModelAdapter';
 
 export class ElasticSearchOrdersAdapter implements ElasticSearchModelAdapter {
-  public readonly model = models.Order;
+  public readonly model = Order;
   public readonly index = ElasticSearchIndexName.ORDERS;
   public readonly mappings = {
     properties: {
@@ -33,7 +33,7 @@ export class ElasticSearchOrdersAdapter implements ElasticSearchModelAdapter {
       ids?: number[];
     } = {},
   ) {
-    return models.Order.findAll({
+    return Order.findAll({
       attributes: omit(Object.keys(this.mappings.properties), ['HostCollectiveId', 'ParentCollectiveId']),
       order: [['id', 'DESC']],
       limit: options.limit,
@@ -47,14 +47,14 @@ export class ElasticSearchOrdersAdapter implements ElasticSearchModelAdapter {
         {
           association: 'collective',
           required: true,
-          attributes: ['HostCollectiveId', 'ParentCollectiveId'],
+          attributes: ['isActive', 'HostCollectiveId', 'ParentCollectiveId'],
         },
       ],
     });
   }
 
   public mapModelInstanceToDocument(
-    instance: InstanceType<typeof models.Order>,
+    instance: InstanceType<typeof Order>,
   ): Record<keyof (typeof this.mappings)['properties'], unknown> {
     return {
       id: instance.id,
@@ -63,7 +63,7 @@ export class ElasticSearchOrdersAdapter implements ElasticSearchModelAdapter {
       description: instance.description,
       CollectiveId: instance.CollectiveId,
       FromCollectiveId: instance.FromCollectiveId,
-      HostCollectiveId: instance.collective.HostCollectiveId,
+      HostCollectiveId: !instance.collective.isActive ? null : instance.collective.HostCollectiveId,
       ParentCollectiveId: instance.collective.ParentCollectiveId,
     };
   }