Merge pull request #24 from demokratie-live/sprint#6/new_webhook

Sprint#6/new webhook

Author: ManAnRuck

scripts/importAll.js

@@ -7,13 +7,11 @@ import getAllProcedures from '../src/graphql/queries/getAllProcedures';
 
 require('../src/config/db');
 
-const PAGE_SIZE = 20;
-
 (async () => {
   console.log('Start Importing');
   const { data: { allProcedures } } = await client.query({
     query: getAllProcedures,
-    variables: { pageSize: PAGE_SIZE },
+    // variables: {},
   });
   console.log(allProcedures.map(({ procedureId }) => procedureId));
   console.log('Start Inserting');

src/graphql/queries/getAllProcedures.js

@@ -1,8 +1,8 @@
 import gql from 'graphql-tag';
 
 export default gql`
-  query allProcedures($pageSize: Int) {
-    allProcedures(pageSize: $pageSize) {
+  query allProcedures() {
+    allProcedures() {
       title
       procedureId
       type

src/graphql/queries/getProcedureUpdates.js

@@ -0,0 +1,10 @@
+import gql from 'graphql-tag';
+
+export default gql`
+  query procedureUpdates($period: [Int!], $type: [String!]) {
+    procedureUpdates(period: $period, type: $type) {
+      procedureId
+      updatedAt
+    }
+  }
+`;

src/graphql/queries/getProcedures.js

@@ -1,8 +1,8 @@
 import gql from 'graphql-tag';
 
 export default gql`
-  query procedures($pageSize: Int, $IDs: [String!]) {
-    procedures(pageSize: $pageSize, IDs: $IDs) {
+  query procedures($IDs: [String!]) {
+    procedures(IDs: $IDs) {
       title
       procedureId
       type

src/index.js

@@ -12,7 +12,7 @@ import constants from './config/constants';
 import typeDefs from './graphql/schemas';
 import resolvers from './graphql/resolvers';
 
-import importProcedures from './scripts/import';
+import webhook from './scripts/webhook';
 
 // Models
 import ProcedureModel from './models/Procedure';
@@ -57,13 +57,16 @@ app.use(constants.GRAPHQL_PATH, (req, res, next) => {
 });
 
 app.post('/webhooks/bundestagio/update', async (req, res) => {
-  const { procedureIds } = req.body;
+  const { data } = req.body;
   try {
+    const updated = await webhook(data);
     res.send({
-      updated: await importProcedures(procedureIds),
+      updated,
       succeeded: true,
     });
+    console.log(`Updated: ${updated}`);
   } catch (error) {
+    console.log(error);
     res.send({
       error,
       succeeded: false,

src/scripts/import.js

@@ -5,13 +5,11 @@ import client from '../graphql/client';
 import Procedure from '../models/Procedure';
 import getProcedures from '../graphql/queries/getProcedures';
 
-const PAGE_SIZE = 20;
-
 export default async (procedureIds) => {
   // Start Importing
   const { data: { procedures } } = await client.query({
     query: getProcedures,
-    variables: { pageSize: PAGE_SIZE, IDs: procedureIds },
+    variables: { IDs: procedureIds },
   });
   // Start Inserting
   const promises = await procedures.map(async (bIoProcedure) => {

src/scripts/webhook.js

@@ -0,0 +1,66 @@
+import importProcedures from './import';
+import getProcedureUpdates from '../graphql/queries/getProcedureUpdates';
+import client from '../graphql/client';
+import ProcedureModel from '../models/Procedure';
+
+export default async (data) => {
+  // Count local Data in groups
+  const groups = await ProcedureModel.aggregate([{
+    // Group by Period & Type
+    $group: {
+      _id: { period: '$period', type: '$type' },
+      count: { $sum: 1 },
+    },
+  },
+  {
+    // Group by Period
+    $group: {
+      _id: '$_id.period',
+      types: { $push: { type: '$_id.type', count: '$count' } },
+    },
+  },
+  {
+    // Rename _id Field to period
+    $project: { _id: 0, period: '$_id', types: 1 },
+  }]);
+
+  const update = [];
+  await Promise.all(data.map(async (d) => {
+    const period = parseInt(d.period, 10);
+    const { type, countBefore, changedIds } = d.types.find(t => t.type === 'Gesetzgebung');
+    const group = groups.find(c => c.period === period);
+    const localCount = group ? group.types
+      .find(ct => ct.type === type).count : 0;
+    // Append Changed IDs
+    update.concat(changedIds);
+    // Compare Counts Remote & Local
+    if (countBefore > localCount) {
+      // Find remote Procedure Updates
+      const { data: { procedureUpdates } } = await client.query({
+        query: getProcedureUpdates,
+        variables: { period, type },
+      });
+      // Find local Procedure Updates
+      const localProcedureUpdates = await ProcedureModel
+        .find({ period, type }, { procedureId: 1, lastUpdateDate: 1 });
+      // Compare
+      procedureUpdates.forEach((pu) => {
+        const localData = localProcedureUpdates.find(ld => ld.procedureId === pu.procedureId);
+        if (!localData || new Date(localData.lastUpdateDate) < new Date(pu.updatedAt)) {
+          update.push(pu.procedureId);
+        }
+      });
+    }
+  }));
+
+  // Splitt in Chunks & Update
+  const chunkSize = 100;
+  let updateCount = 0;
+  let i = 0;
+  for (i = 0; i < update.length; i += chunkSize) {
+    const part = update.slice(i, i + chunkSize);
+    updateCount += await importProcedures(part); // eslint-disable-line no-await-in-loop
+  }
+
+  return updateCount;
+};