Sync Architecture Patterns

async function pollAllResources(customerId) {
  await pollUsers(customerId);
  await pollProjects(customerId);
  await pollGroups(customerId);
  await pollForms(customerId);
  await pollFormCompletions(customerId);
  // ... etc
}

async function pushFromExternalToVomo(customerId, externalUpdate) {
  // 1. Check if we just received this change from VOMO direction
  const recentInbound = await db.getRecentInboundChange(
    customerId,
    externalUpdate.vomoUserId,
    /* withinSeconds */ 60
  );

  if (recentInbound) {
    return { skipped: true, reason: 'recent_inbound_from_vomo' };
  }

  // 2. Push to VOMO
  await pushToVomo(customerId, externalUpdate);

  // 3. Track that this was an outbound change
  await db.recordOutboundChange(customerId, externalUpdate.vomoUserId);
}

const FIELD_AUTHORITY = {
  // VOMO is canonical for these
  first_name: 'vomo',
  last_name: 'vomo',
  email: 'vomo',

  // External (e.g., HR system) is canonical for these
  phone: 'external',
  address: 'external',
  birthday: 'external',
};

function shouldPushFieldToVomo(fieldName) {
  return FIELD_AUTHORITY[fieldName] === 'external';
}

async function scheduleAllCustomerPolls() {
  const customers = await db.getActiveCustomers();
  const pollIntervalMs = 15 * 60 * 1000; // 15 min

  for (let i = 0; i < customers.length; i++) {
    const offsetMs = (i / customers.length) * pollIntervalMs;
    scheduler.cron(
      `poll:${customers[i].id}`,
      `*/15 * * * *`, // every 15 min
      () => pollCustomer(customers[i].id),
      { initialDelayMs: offsetMs } // stagger initial start
    );
  }
}

class PerCustomerBudget {
  async checkAndDeduct(customerId, requestCount) {
    const minute = Math.floor(Date.now() / 60000);
    const key = `budget:${customerId}:${minute}`;
    const current = await redis.incrby(key, requestCount);
    await redis.expire(key, 120);

    const limit = 100; // per minute per customer
    return current <= limit;
  }
}

class TieredScheduler {
  async scheduleCustomer(customerId) {
    const tier = await getCustomerTier(customerId);

    switch (tier) {
      case 'premium':
        return scheduler.cron(`poll:${customerId}`, '*/5 * * * *',
          () => pollCustomer(customerId), { lane: 'priority' });
      case 'standard':
        return scheduler.cron(`poll:${customerId}`, '*/15 * * * *',
          () => pollCustomer(customerId), { lane: 'standard' });
      case 'basic':
        return scheduler.cron(`poll:${customerId}`, '0 * * * *',
          () => pollCustomer(customerId), { lane: 'background' });
    }
  }
}

async function pollUsers(customerId) {
  const useNewLogic = await featureFlags.isEnabled(
    'new_user_polling_logic',
    { customerId }
  );

  if (useNewLogic) {
    return pollUsersV2(customerId);
  }
  return pollUsersV1(customerId);
}

// Polling layer — knows about VOMO, checkpoints, pagination
class UserPoller {
  async poll(customerId) {
    const checkpoint = await checkpoints.get(customerId, 'users');
    const changes = await listChangedUsers(customerId, checkpoint);

    for (const change of changes) {
      await queue.publish({
        type: 'user_changed',
        customerId,
        userData: change,
      });
    }

    const latest = latestUpdatedAt(changes) ?? checkpoint;
    await checkpoints.set(customerId, 'users', latest);
  }
}

// Business layer — processes events, knows about destination
class UserChangeHandler {
  async handle(event) {
    const customerId = event.customerId;
    const user = event.userData;
    // Business decisions — mapping, transformation, side effects
    await externalSystem.upsertUser(transformToExternal(user));
    await fireBusinessEvents(customerId, user);
  }
}