todos-contra-el-fuego-web/imports/startup/server/subsUnion.js

/* eslint-disable import/no-absolute-path */

import { Meteor } from 'meteor/meteor';
import Subscriptions from '/imports/api/Subscriptions/Subscriptions';
import SiteSettings from '/imports/api/SiteSettings/SiteSettings';
import Perlin from 'loms.perlin';
import './leaflet-workaround';
import L from 'leaflet';
import calcUnionAsync from '/imports/startup/server/calcUnionAsync';
import { isMailServerMaster } from '/imports/startup/server/email';

// sudo apt-get install libcairo2-dev libjpeg-dev libgif-dev

Meteor.startup(async () => {
  if (!isMailServerMaster) {
    console.log('We only process subsUnion in master');
    return;
  }
  const debug = true; // Meteor.isDevelopment;
  Perlin.seed(Math.random());

  const addNoisy = (osub) => {
    const sub = osub;
    let lat = Math.round(sub.location.lat * 10) / 10;
    let lon = Math.round(sub.location.lon * 10) / 10;
    const noiseBase = Perlin.perlin2(lat, lon);
    const noise = Math.abs(noiseBase / 3);
    lat += noise;
    lon += noise;
    sub.location.lat = lat;
    sub.location.lon = lon;
    sub.distance += noiseBase;
    return sub;
  };

  const noNoisy = sub => sub;

  const process = async (isPublic) => {
    const subscribers = await Subscriptions.find().fetchAsync();
    const decorate = isPublic ? addNoisy : noNoisy;
    const validSubs = subscribers.filter((osub) => {
      const valid = osub.location && osub.location.lat && osub.location.lon && osub.distance;
      if (!valid) console.info(`Wrong element to do union ${JSON.stringify(osub)}`);
      return valid;
    });
    const decorated = validSubs.map(decorate);

    let union = null;
    try {
      union = await calcUnionAsync(decorated);
    } catch (e) {
      console.error('subsUnion worker failed', e);
    }
    const bounds = union === null ? null : L.geoJSON(union).getBounds();

    const publicl = isPublic ? 'public' : 'private';
    const Publicl = publicl.replace(/\b\w/g, l => l.toUpperCase());

    if (typeof union === 'object') {
      const unionSet = {
        $set: {
          name: `subs-${publicl}-union`,
          value: JSON.stringify(union),
          isPublic,
          description: `${Publicl} subscriptions union`,
          type: 'string'
        }
      };
      const boundsSet = {
        $set: {
          name: `subs-${publicl}-union-bounds`,
          value: JSON.stringify(bounds),
          isPublic,
          description: `${Publicl} subscriptions union bounds`,
          type: 'string'
        }
      };
      const sizeSet = {
        $set: {
          name: 'subs-union-count',
          value: subscribers.length,
          isPublic: false,
          description: 'Subscriptions count',
          type: 'number'
        }
      };
      // FIXME, take care of object size:
      // https://stackoverflow.com/questions/10827812/what-is-the-length-maximum-for-a-string-data-type-in-mongodb-used-with-ruby
      await SiteSettings.upsertAsync({ name: `subs-${publicl}-union` }, unionSet, { multi: false });
      await SiteSettings.upsertAsync({ name: `subs-${publicl}-union-bounds` }, boundsSet, { multi: false });
      await SiteSettings.upsertAsync({ name: 'subs-union-count' }, sizeSet, { multi: false });
      if (debug) console.log(`${Publicl} subscription union calculated`);
    } else {
      console.log('Subscription union failed!');
    }
  };

  const recreate = async () => { await process(true); await process(false); };

  let recomputeRunning = false;
  let recomputePending = false;

  // Runs recreate() serially: a recompute already in flight is never overlapped by
  // another one, bursts of subscription changes just mark it pending and get a single
  // extra pass once the current one finishes.
  const runRecreate = async () => {
    recomputeRunning = true;
    try {
      await recreate();
    } catch (e) {
      console.error('subsUnion recompute failed', e);
    } finally {
      recomputeRunning = false;
      if (recomputePending) {
        recomputePending = false;
        runRecreate();
      }
    }
  };

  // Fire-and-forget on purpose: recreate() can take a while over thousands of
  // subscriptions, and callers (Meteor.startup, observeAsync callbacks) must not
  // block on it.
  const scheduleRecreate = () => {
    if (recomputeRunning) {
      recomputePending = true;
      return;
    }
    runRecreate();
  };

  // At startup, we check if it's necessary to calc subscriptions union again
  const currentUnion = await SiteSettings.findOneAsync({ name: 'subs-public-union' });
  const lastSubs = await Subscriptions.findOneAsync({}, { sort: { updatedAt: -1 } });
  const countUnionSubs = await SiteSettings.findOneAsync({ name: 'subs-union-count' });
  const countSubs = await Subscriptions.find({}).countAsync();
  if (currentUnion && lastSubs) {
    const lastUnionUpdated = currentUnion.updatedAt;
    const lastSubsUpdated = lastSubs.updatedAt;
    if (lastUnionUpdated > lastSubsUpdated || !countUnionSubs || countSubs !== countUnionSubs.value) {
      console.log('Subs union outdated');
      scheduleRecreate();
    } else {
      console.log('Subs union up-to-date');
    }
  }

  await Subscriptions.find({ createdAt: { $gt: new Date() } }).observeAsync({
    added: function newSubAdded() { // doc) {
      if (debug) console.log('Subs added so recreate union');
      scheduleRecreate();
    }
  });

  await Subscriptions.find().observeAsync({
    changed: function subsChanged() { // updatedDoc, oldDoc) {
      if (debug) console.log('Subs changed so recreate union');
      scheduleRecreate();
    },
    removed: function subsRemoved() { // oldDoc) {
      if (debug) console.log('Subs removed so recreate union');
      scheduleRecreate();
    }
  });
});