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

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

// All of the subscriptions-union behaviour, with its collaborators injected.
//
// This used to live inside the Meteor.startup() callback of subsUnion.js, where
// nothing was reachable from a test: the queue, the incremental fast path and the
// worker-failure handling could only be exercised by starting a whole server and
// hoping. That is how a failed worker call ended up being stored as a valid empty
// union (50ca9cc) and how a full recompute ended up blocking DDP (b4e5511 /
// fbb746f) — three production incidents in code with zero test coverage.
// subsUnion.js is now only the wiring; everything below is a plain function you
// can call with fakes.

import Perlin from 'loms.perlin';
import limitUnionSize, { MAX_UNION_BYTES } from './unionSizeGuard';

// Telegram-era subscriptions (and anything else half-written) can be missing the
// location or the radius. They are skipped, not an error: they predate the web
// subscriptions and there is nothing sensible to draw for them.
export const isValidSub = (osub, log = console) => {
  const valid = !!(osub && osub.location && osub.location.lat && osub.location.lon && osub.distance);
  if (!valid) log.info(`Wrong element to do union ${JSON.stringify(osub)}`);
  return valid;
};

/**
 * @param {object} deps
 * @param {function} deps.calcUnion   (subs, baseUnion) => Promise<union|null>, off-thread
 * @param {object}   deps.SiteSettings Mongo collection holding the stored union
 * @param {object}   deps.Subscriptions Mongo collection of subscriptions
 * @param {function} [deps.boundsOf]  union => leaflet-style bounds (or null)
 * @param {number}   [deps.maxUnionBytes]
 * @param {number}   [deps.perlinSeed] fixed seed makes the public "noise" reproducible in tests
 */
const noTelemetry = { computed: () => {}, failed: () => {} };

const createSubsUnion = ({
  calcUnion,
  SiteSettings,
  Subscriptions,
  boundsOf = () => null,
  maxUnionBytes = MAX_UNION_BYTES,
  perlinSeed = Math.random(),
  debug = true,
  log = console,
  telemetry = noTelemetry
}) => {
  Perlin.seed(perlinSeed);

  // Public unions are fuzzed so that the map does not disclose where exactly
  // somebody subscribed. Returns a new object: callers hand us documents straight
  // out of Mongo and the private pass must not see the fuzzed values.
  const addNoisy = (sub) => {
    const lat = Math.round(sub.location.lat * 10) / 10;
    const lon = Math.round(sub.location.lon * 10) / 10;
    const noiseBase = Perlin.perlin2(lat, lon);
    const noise = Math.abs(noiseBase / 3);
    return {
      ...sub,
      location: { ...sub.location, lat: lat + noise, lon: lon + noise },
      distance: sub.distance + noiseBase
    };
  };

  const noNoisy = sub => ({ ...sub, location: { ...sub.location } });

  const decoratorFor = isPublic => (isPublic ? addNoisy : noNoisy);

  // Only called once calcUnion 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 calcUnion failures themselves and skip calling this.)
  const storeUnion = async (isPublic, union, count) => {
    const publicl = isPublic ? 'public' : 'private';
    const Publicl = publicl.replace(/\b\w/g, l => l.toUpperCase());
    // Mongo caps a document at 16 MiB and this value is also pushed to every
    // browser: past a certain number of subscriptions the geometry has to be
    // coarsened or the upsert simply starts failing. See unionSizeGuard.js.
    const { union: safeUnion, json, degraded } = limitUnionSize(union, maxUnionBytes);
    if (degraded) {
      log.warn(`${Publicl} subscriptions union too big (${degraded.originalBytes} bytes), degraded via ${degraded.steps.join(' -> ')}`);
    }
    const bounds = boundsOf(safeUnion);
    const unionSet = {
      $set: {
        name: `subs-${publicl}-union`,
        value: json,
        isPublic,
        description: `${Publicl} subscriptions union`,
        type: 'string',
        degraded: degraded ? degraded.steps.join(' -> ') : null
      }
    };
    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'
      }
    };
    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) log.log(`${Publicl} subscription union calculated`);
    return { degraded, bytes: Buffer.byteLength(json, 'utf8') };
  };

  const recomputeOne = async (isPublic) => {
    const subscribers = await Subscriptions.find().fetchAsync();
    const decorated = subscribers.filter(s => isValidSub(s, log)).map(decoratorFor(isPublic));

    const started = Date.now();
    try {
      const union = await calcUnion(decorated);
      const { degraded, bytes } = await storeUnion(isPublic, union, subscribers.length);
      telemetry.computed({
        kind: 'recreate',
        isPublic,
        subs: decorated.length,
        ms: Date.now() - started,
        bytes,
        degraded: degraded ? degraded.steps.join(' -> ') : null
      });
    } catch (e) {
      log.error('subsUnion worker failed, union left unchanged', e);
      telemetry.failed(e, { kind: 'recreate', isPublic, subs: decorated.length, ms: Date.now() - started });
    }
  };

  const recreate = async () => { await recomputeOne(true); await recomputeOne(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;
  let idleWaiters = [];

  const flushIdleWaiters = () => {
    const waiters = idleWaiters;
    idleWaiters = [];
    waiters.forEach(resolve => resolve());
  };

  // 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, log)) return;

    const countSetting = await SiteSettings.findOneAsync({ name: 'subs-union-count' });
    const countSubs = await Subscriptions.find({}).countAsync();
    if (!countSetting || countSetting.value !== countSubs - 1) {
      if (debug) log.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 decorated = decoratorFor(isPublic)(doc);
      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;

      const started = Date.now();
      try {
        // eslint-disable-next-line no-await-in-loop
        const union = await calcUnion([decorated], baseUnion);
        // eslint-disable-next-line no-await-in-loop
        const { degraded, bytes } = await storeUnion(isPublic, union, countSubs);
        telemetry.computed({
          kind: 'incrementalAdd',
          isPublic,
          subs: countSubs,
          ms: Date.now() - started,
          bytes,
          degraded: degraded ? degraded.steps.join(' -> ') : null
        });
      } catch (e) {
        log.error('subsUnion incremental worker failed, falling back to a full recreate', e);
        telemetry.failed(e, { kind: 'incrementalAdd', isPublic, subs: countSubs, ms: Date.now() - started });
        recomputePending = true;
        return;
      }
    }
    if (debug) log.log('Subs union updated incrementally');
  };

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

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

  // Resolves when the queue has drained and nothing is running. Only meant for
  // tests and shutdown — production code must never await the union.
  const idle = () => new Promise((resolve) => {
    if (!busy && addQueue.length === 0 && !recomputePending) {
      resolve();
      return;
    }
    idleWaiters.push(resolve);
  });

  // At startup we check whether the stored union still matches the subscriptions.
  const isUnionOutdated = async () => {
    const currentUnion = await SiteSettings.findOneAsync({ name: 'subs-public-union' });
    const lastSubs = await Subscriptions.findOneAsync({}, { sort: { updatedAt: -1 } });
    if (!currentUnion || !lastSubs) return false;
    const countUnionSubs = await SiteSettings.findOneAsync({ name: 'subs-union-count' });
    const countSubs = await Subscriptions.find({}).countAsync();
    return currentUnion.updatedAt > lastSubs.updatedAt
      || !countUnionSubs
      || countSubs !== countUnionSubs.value;
  };

  return {
    addNoisy,
    noNoisy,
    storeUnion,
    recreate,
    incrementalAdd,
    enqueueAdd,
    scheduleRecreate,
    isUnionOutdated,
    idle,
    stats: () => ({ busy, queued: addQueue.length, recomputePending })
  };
};

export default createSubsUnion;