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 isValidSub = (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;
  };

  // Only called once calcUnionAsync has actually resolved — union === null here
  // means "zero valid subscriptions", a real result, not a failed computation.
  // (A prior version guarded this with `typeof union === 'object'`, which is
  // true for null too, so a failed worker call — caught upstream, leaving
  // union as null — looked identical to a legitimate empty union and got
  // silently stored as "null". Callers must catch calcUnionAsync failures
  // themselves and skip calling this, instead of relying on this check.)
  const storeUnion = async (isPublic, union, count) => {
    const publicl = isPublic ? 'public' : 'private';
    const Publicl = publicl.replace(/\b\w/g, l => l.toUpperCase());
    const bounds = union === null ? null : L.geoJSON(union).getBounds();
    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: count,
        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`);
  };

  const process = async (isPublic) => {
    const subscribers = await Subscriptions.find().fetchAsync();
    const decorate = isPublic ? addNoisy : noNoisy;
    const decorated = subscribers.filter(isValidSub).map(decorate);

    try {
      const union = await calcUnionAsync(decorated);
      await storeUnion(isPublic, union, subscribers.length);
    } catch (e) {
      console.error('subsUnion worker failed, union left unchanged', e);
    }
  };

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

  // Single serialized worker over two kinds of work: incremental adds (a real
  // FIFO queue — each must see the previous one's result, none get dropped)
  // and full recomputes (coalesced — a burst of changed/removed events only
  // needs one pass once things settle). Both touch the same stored union, so
  // they share this one lock rather than racing each other.
  let busy = false;
  const addQueue = [];
  let recomputePending = false;

  // Fast path for the common case (a new subscription): merge just the new
  // circle into the union already stored instead of redoing the full turf.union
  // chain over every subscription again (that chain is what made a single
  // recompute take ~20+ minutes with thousands of subscriptions). Falls back to
  // a full recreate() if the stored state isn't exactly "one behind" what we
  // expect — safer than merging into stale/inconsistent data. Sets
  // recomputePending directly (rather than calling a schedule helper) since
  // runLoop already re-checks it right after this returns.
  const incrementalAdd = async (doc) => {
    if (!isValidSub(doc)) return;

    const countSetting = await SiteSettings.findOneAsync({ name: 'subs-union-count' });
    const countSubs = await Subscriptions.find({}).countAsync();
    if (!countSetting || countSetting.value !== countSubs - 1) {
      if (debug) console.log('Subs union state not exactly one behind, doing a full recreate instead');
      recomputePending = true;
      return;
    }

    // eslint-disable-next-line no-restricted-syntax
    for (const isPublic of [true, false]) {
      const decorate = isPublic ? addNoisy : noNoisy;
      const decorated = decorate({ ...doc, location: { ...doc.location } });
      const publicl = isPublic ? 'public' : 'private';
      // eslint-disable-next-line no-await-in-loop
      const current = await SiteSettings.findOneAsync({ name: `subs-${publicl}-union` });
      const baseUnion = current && current.value ? JSON.parse(current.value) : null;

      try {
        // eslint-disable-next-line no-await-in-loop
        const union = await calcUnionAsync([decorated], baseUnion);
        // eslint-disable-next-line no-await-in-loop
        await storeUnion(isPublic, union, countSubs);
      } catch (e) {
        console.error('subsUnion incremental worker failed, falling back to a full recreate', e);
        recomputePending = true;
        return;
      }
    }
    if (debug) console.log('Subs union updated incrementally');
  };

  const runLoop = () => {
    if (busy) return;
    if (addQueue.length > 0) {
      busy = true;
      const doc = addQueue.shift();
      incrementalAdd(doc)
        .catch(e => console.error('subsUnion incremental add failed', e))
        .finally(() => { busy = false; runLoop(); });
      return;
    }
    if (recomputePending) {
      recomputePending = false;
      busy = true;
      recreate()
        .catch(e => console.error('subsUnion recompute failed', e))
        .finally(() => { busy = false; runLoop(); });
    }
  };

  const enqueueAdd = (doc) => { addQueue.push(doc); runLoop(); };

  // Fire-and-forget on purpose: both recreate() and incrementalAdd() can take a
  // while, and callers (Meteor.startup, observeAsync callbacks) must not block on it.
  const scheduleRecreate = () => { recomputePending = true; runLoop(); };

  // 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, merging into union incrementally');
      enqueueAdd(doc);
    }
  });

  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();
    }
  });
});