server/laf-cloud/tests/recharge-stats.test.mjs

import test from 'node:test';
import assert from 'node:assert/strict';
import { registerHooks, stripTypeScriptTypes } from 'node:module';
import { readFileSync } from 'node:fs';

const DAY = 86400000;
const OFFSET = 8 * 3600000;
const NOW = Date.parse('2026-09-15T03:00:00+08:00');
const CUTOFF = Date.parse('2026-09-14T23:59:59.999+08:00');
const state = { users: [], order: [], userRechargeStats: [], reads: [], orderQueries: [], writes: 0, closed: 0, beforeWrite: null, beforeUserWrite: null, failUserWrite: false, vipWrites: 0, failRead: false };
const realNow = Date.now;
Date.now = () => NOW;
test.after(() => { Date.now = realNow; });

function matches(row, query) {
  return Object.entries(query).every(([key, expected]) => {
    if (key === '$or') return expected.some(part => matches(row, part));
    const actual = key.split('.').reduce((value, part) => value?.[part], row);
    if (expected === null) return actual == null;
    if (typeof expected !== 'object') return actual === expected;
    return Object.entries(expected).every(([op, value]) => {
      if (op === '$in') return value.includes(actual);
      if (op === '$ne') return actual !== value;
      if (op === '$gt') return actual > value;
      if (op === '$lte') return actual <= value;
      if (op === '$exists') return (actual !== undefined) === value;
      if (op === '$not') return !value.test(actual);
      throw new Error(`Unsupported query operator: ${op}`);
    });
  });
}

function evaluate(expression, row) {
  if (expression === '$$ROOT') return row;
  if (typeof expression === 'string' && expression.startsWith('$')) return expression.slice(1).split('.').reduce((value, key) => value?.[key], row);
  if (!expression || typeof expression !== 'object') return expression;
  if ('$literal' in expression) return expression.$literal;
  if (!Object.keys(expression)[0]?.startsWith('$')) return Object.fromEntries(
    Object.entries(expression).map(([key, value]) => [key, evaluate(value, row)]));
  const [operator, values] = Object.entries(expression)[0];
  if (operator === '$mergeObjects') return Object.assign({}, ...values.map(value => evaluate(value, row)));
  if (operator === '$cond') return evaluate(values[evaluate(values[0], row) ? 1 : 2], row);
  if (operator === '$or') return values.some(value => evaluate(value, row));
  if (operator === '$ne') return evaluate(values[0], row) !== evaluate(values[1], row);
  if (operator === '$gt') return evaluate(values[0], row) > evaluate(values[1], row);
  if (operator === '$and') return values.every(value => evaluate(value, row));
  if (operator === '$ifNull') return evaluate(values[0], row) ?? evaluate(values[1], row);
  if (operator === '$lte') return evaluate(values[0], row) <= evaluate(values[1], row);
  if (operator === '$eq') {
    const a = evaluate(values[0], row), b = evaluate(values[1], row);
    return a instanceof Date && b instanceof Date ? a.getTime() === b.getTime() : a === b;
  }
  throw new Error(`Unsupported update expression: ${operator}`);
}

globalThis.__rechargeStatsCloud = { mongo: { db: {
  collection(name) {
    assert.ok(['users', 'order', 'userRechargeStats'].includes(name), 'must not count iosOrder temporary records');
    return {
      find(query, options) {
        state.reads.push(name);
        if (name === 'order') state.orderQueries.push(structuredClone(query));
        if (name === 'users') assert.deepEqual(options.projection, { _id: 1, openid: 1, pay_user: 1, register_time: 1, vip_level: 1, vip_profile: 1 });
        let rows = state[name].filter(row => matches(row, query));
        return {
          sort(sort) {
            rows.sort((a, b) => {
              for (const key of Object.keys(sort)) {
                if (a[key] < b[key]) return -1;
                if (a[key] > b[key]) return 1;
              }
              return 0;
            });
            return this;
          },
          limit(size) { rows = rows.slice(0, size); return this; },
          async toArray() { return rows.map(row => Object.fromEntries(
            Object.keys(options.projection).filter(key => row[key] !== undefined).map(key => [key, row[key]])
          )); },
          async *[Symbol.asyncIterator]() {
            if (state.failRead) throw new Error('database read failed');
            yield* rows;
          },
          async close() { state.closed++; },
        };
      },
      async bulkWrite(operations) {
        assert.notEqual(name, 'order', 'orders remain read-only');
        if (name === 'userRechargeStats') { state.writes++; state.beforeWrite?.(); }
        else {
          state.vipWrites++; state.beforeUserWrite?.();
          if (state.failUserWrite) throw new Error('users write failed');
        }
        let modifiedCount = 0;
        let upsertedCount = 0;
        for (const { updateOne } of operations) {
          assert.deepEqual(Object.keys(updateOne.filter), ['_id']);
          assert.equal(updateOne.upsert, name === 'userRechargeStats');
          assert.equal(updateOne.update.length, 1);
          const index = state[name].findIndex(row => matches(row, updateOne.filter));
          if (index < 0 && !updateOne.upsert) continue;
          const previous = index < 0 ? { _id: updateOne.filter._id } : state[name][index];
          const next = structuredClone(evaluate(updateOne.update[0].$replaceWith, previous));
          if (index < 0) { state[name].push(next); upsertedCount++; }
          else if (JSON.stringify(previous) !== JSON.stringify(next)) {
            state[name][index] = next; modifiedCount++;
          }
        }
        return { modifiedCount, upsertedCount };
      },
    };
  },
} } };
registerHooks({ resolve(specifier, context, next) {
  if (specifier === '@lafjs/cloud') return {
    url: 'data:text/javascript,export default globalThis.__rechargeStatsCloud', shortCircuit: true,
  };
  if (specifier === '@/paymentProfile') return {
    url: new URL('../functions/paymentProfile.ts', import.meta.url).href, shortCircuit: true,
  };
  return next(specifier, context);
}, load(url, context, next) {
  if (['rechargeStats', 'paymentProfile'].some(name => url === new URL(`../functions/${name}.ts`, import.meta.url).href)) {
    return { format: 'module', source: stripTypeScriptTypes(readFileSync(new URL(url), 'utf8')), shortCircuit: true };
  }
  return next(url, context);
} });
const { default: run } = await import('../functions/rechargeStats.ts');

function reset() {
  Object.assign(state, {
    users: [{ _id: 'u1', openid: 'o1', pay_user: true }], order: [], userRechargeStats: [], reads: [], orderQueries: [],
    writes: 0, closed: 0, beforeWrite: null, beforeUserWrite: null, failUserWrite: false, vipWrites: 0, failRead: false,
  });
}
function userFor(id = 'u1') { return state.users.find(row => row._id === id); }
function withoutVip(user) { const { vip_level, vip_profile, ...rest } = user; return rest; }
function statsFor(id = 'u1') { return state.userRechargeStats.find(row => row._id === id); }
function order(id, age, extra = {}) {
  return {
    _id: id, outTradeNo: id, openid: 'o1', state: 2, goodsPrice: 100, itemCount: 1,
    chargeTime: new Date(CUTOFF - age + OFFSET), time: CUTOFF - age - DAY, ...extra,
  };
}

test('windows exclude the previous end-of-day boundary and include yesterday final millisecond', async () => {
  reset();
  state.order = [
    order('yesterday-end', 0), order('15d-start', 15 * DAY - 1), order('15d-excluded', 15 * DAY),
    order('30d-start', 30 * DAY - 1), order('30d-excluded', 30 * DAY), order('today-start', -1),
    order('early-morning', -2 * 3600000), order('future', -4 * 3600000),
  ];
  await run();
  const stats = statsFor();
  assert.equal(stats.amount15d, 200);
  assert.equal(stats.amount30d, 400);
  assert.equal(stats.amountTotal, 500);
  assert.equal(stats.orderCount, 5);
  assert.equal(stats.asOf, CUTOFF);
  assert.equal(stats.version, 8);
  assert.equal(stats.unit, 'fen');
  assert.equal(stats.updatedAt, NOW);
  assert.equal(stats._id, 'u1');
  assert.equal(stats.openid, 'o1');
  assert.equal(state.users[0].rechargeStats, undefined);
});

test('counts quantities, numeric strings, paid states and deduplicates; excludes unpaid and test orders', async () => {
  reset();
  state.order = [
    order('a', DAY, { goodsPrice: '101', itemCount: '3', state: 1 }),
    order('b', DAY, { goodsPrice: 7, itemCount: 1 }),
    order('duplicate', DAY, { outTradeNo: 'a', goodsPrice: 101, itemCount: 3 }),
    order('pending', DAY, { state: 0 }), order('string-state', DAY, { state: '2' }),
    order('test-env', DAY, { paymentAppEnv: 'test' }), order('wct_test', DAY),
  ];
  await run();
  assert.equal(statsFor().amountTotal, 310);
  assert.equal(statsFor().orderCount, 2);
  assert.equal(statsFor().duplicateOrderCount, 1);
});

test('falls back to unshifted creation time, tracks unknown times, and rejects malformed amounts', async () => {
  reset();
  state.order = [
    order('fallback', 0, { chargeTime: 0, time: CUTOFF - 15 * DAY + 1 }),
    order('unknown', 0, { chargeTime: 0, time: null }),
    order('iso', 0, { chargeTime: new Date(CUTOFF - 30 * DAY + 1 + OFFSET).toISOString() }),
    order('bad-price', 0, { goodsPrice: 'bad' }), order('bad-count', 0, { itemCount: null }),
    order('negative', 0, { goodsPrice: -1 }), order('missing-id', 0, { outTradeNo: '' }),
    order('overflow', 0, { goodsPrice: Number.MAX_SAFE_INTEGER, itemCount: 2 }),
  ];
  await run();
  const stats = statsFor();
  assert.equal(stats.amount15d, 100);
  assert.equal(stats.amount30d, 200);
  assert.equal(stats.amountTotal, 300);
  assert.equal(stats.fallbackTimeOrderCount, 1);
  assert.equal(stats.missingTimeOrderCount, 1);
  assert.equal(stats.invalidOrderCount, 5);
});

test('maximum uses a whole eligible historical order, with the same cutoff and deduplication rules', async () => {
  reset();
  state.order = [
    order('old', 60 * DAY, { goodsPrice: '300', itemCount: '3', state: 1 }),
    order('recent', DAY, { goodsPrice: 500 }),
    order('z-duplicate', DAY, { outTradeNo: 'old', goodsPrice: 9999 }),
    order('today', -1, { goodsPrice: 9999 }),
    order('pending', DAY, { state: 0, goodsPrice: 9999 }),
    order('test', DAY, { paymentAppEnv: 'test', goodsPrice: 9999 }),
    order('wct_test', DAY, { goodsPrice: 9999 }),
    order('invalid', DAY, { goodsPrice: 9999, itemCount: null }),
  ];
  await run();
  assert.equal(statsFor().amountMax, 900);
  assert.equal(statsFor().amountTotal, 1400);
  assert.equal(statsFor().amount30d, 500);
  assert.equal(statsFor().unit, 'fen');
});

test('maximum includes undated lifetime orders and is recomputed when orders change or disappear', async () => {
  reset();
  state.userRechargeStats = [{ _id: 'u1', version: 4, asOf: CUTOFF, updatedAt: NOW - 1, amountTotal: 1 }];
  state.order = [order('unknown', DAY, { chargeTime: null, time: null, goodsPrice: 1200 })];
  await run();
  assert.equal(statsFor().amountMax, 1200);
  assert.equal(statsFor().amount30d, 0);
  assert.equal(statsFor().version, 8);
  state.order[0].goodsPrice = 200;
  await run();
  assert.equal(statsFor().amountMax, 200);
  state.order = [];
  await run();
  assert.equal(statsFor().amountMax, 0);
});

test('processes all users across pages, retains legacy totals and distinguishes unknown identity', async () => {
  reset();
  state.users = Array.from({ length: 251 }, (_, i) => ({
    _id: `u${String(i).padStart(4, '0')}`, openid: `o${i}`, pay_user: true,
    rechargeStats: { asOf: NOW - DAY, amountTotal: 999 },
  }));
  state.users.push({ _id: 'no-openid', pay_user: true });
  state.users.push({ _id: 'free', openid: 'free', pay_user: false });
  state.users.push({ _id: 'string', openid: 'string', pay_user: 'true' });
  state.order = [order('last', 1, { openid: 'o250' }), order('free', 1, { openid: 'free' })];
  state.userRechargeStats = [{ _id: 'u0000', asOf: CUTOFF - DAY, amountTotal: 999 }];
  const originalUsers = structuredClone(state.users);
  const result = await run();
  assert.equal(result.data.processedUsers, 254);
  assert.equal(result.data.updatedUsers, 254);
  assert.equal(state.writes, 3);
  assert.equal(statsFor('u0250').amountTotal, 100);
  assert.equal(statsFor('u0250').amountMax, 100);
  assert.equal(statsFor('u0000').amountTotal, 0);
  assert.equal(statsFor('u0000').amountMax, 0);
  assert.equal(statsFor('no-openid').missingOpenid, true);
  assert.equal(statsFor('no-openid').amountMax, 0);
  assert.equal(statsFor('free').data_valid, false, 'paid order contradicts never-paid flag');
  assert.equal(userFor('free').vip_level, null);
  assert.equal(statsFor('string').paid_status, 'unknown');
  assert.equal(userFor('string').vip_level, null);
  assert.equal(state.userRechargeStats.length, 254);
  assert.equal(state.reads.filter(name => name === 'users').length, 4, 'one read per page, not per user');
  assert.deepEqual(state.users.map(withoutVip), originalUsers);
});

test('reruns overwrite instead of accumulating; older runs cannot overwrite newer snapshots', async () => {
  reset();
  state.order = [order('paid', 1)];
  assert.equal((await run()).data.updatedUsers, 1, 'initial upsert counts as an update');
  assert.equal((await run()).data.updatedUsers, 0, 'identical rerun changes nothing');
  assert.equal(statsFor().amountTotal, 100);
  assert.equal(state.userRechargeStats.length, 1);
  state.userRechargeStats[0] = { _id: 'u1', version: 4, asOf: CUTOFF + DAY, updatedAt: NOW + DAY, amountTotal: 200 };
  assert.equal((await run()).data.updatedUsers, 0);
  assert.equal(statsFor().amountTotal, 200);
  assert.equal(state.userRechargeStats.length, 1);
});

test('streams more than 1000 orders and expired windows clear on the next run', async () => {
  reset();
  state.order = Array.from({ length: 1005 }, (_, i) => order(`paid-${i}`, 30 * DAY - 1, { goodsPrice: 1 }));
  await run();
  assert.equal(statsFor().amountTotal, 1005);
  assert.equal(statsFor().amount30d, 1005);
  try {
    Date.now = () => NOW + DAY;
    await run();
    assert.equal(statsFor().amount30d, 0);
    assert.equal(statsFor().amountTotal, 1005);
  } finally { Date.now = () => NOW; }
});

test('a newer same-day snapshot written during calculation is not overwritten or duplicated', async () => {
  reset();
  state.order = [order('paid', DAY)];
  state.beforeWrite = () => { state.userRechargeStats = [
    { _id: 'u1', asOf: CUTOFF, updatedAt: NOW + 60000, amountTotal: 200 },
  ]; };
  assert.equal((await run()).data.updatedUsers, 0);
  assert.equal(statsFor().amountTotal, 200);
  assert.equal(statsFor().updatedAt, NOW + 60000);
  assert.equal(state.userRechargeStats.length, 1);
});

test('an empty eligible set performs no writes and retains previous snapshots', async () => {
  reset();
  state.users = [];
  state.userRechargeStats = [{ _id: 'u1', asOf: CUTOFF - DAY, updatedAt: NOW - DAY, amountTotal: 900 }];
  const previous = structuredClone(state.userRechargeStats);
  const result = await run();
  assert.equal(result.data.processedUsers, 0);
  assert.equal(result.data.updatedUsers, 0);
  assert.equal(state.writes, 0);
  assert.deepEqual(state.reads, ['users']);
  assert.deepEqual(state.userRechargeStats, previous);
});

test('replacement preserves literal identity values and updates an existing same-day snapshot', async () => {
  reset();
  state.users[0].openid = '$openid';
  state.order = [order('paid', DAY, { openid: '$openid' })];
  state.userRechargeStats = [{ _id: 'u1', asOf: CUTOFF, updatedAt: NOW - 60000, amountTotal: 900 }];
  assert.equal((await run()).data.updatedUsers, 1);
  assert.equal(statsFor().openid, '$openid');
  assert.equal(statsFor().amountTotal, 100);
  assert.equal(statsFor().updatedAt, NOW);
  assert.equal(state.userRechargeStats.length, 1);
});

test('read failure propagates, closes cursor and never replaces existing totals with zero', async () => {
  reset();
  state.userRechargeStats = [{ _id: 'u1', asOf: CUTOFF - DAY, updatedAt: NOW - DAY, amountTotal: 9 }];
  state.failRead = true;
  await assert.rejects(run(), /database read failed/);
  assert.equal(state.closed, 1);
  assert.equal(state.writes, 0);
  assert.equal(statsFor().amountTotal, 9);
});

test('timer function does not expose a public HTTP endpoint', () => {
  const config = readFileSync(new URL('../functions/rechargeStats.yaml', import.meta.url), 'utf8');
  assert.match(config, /methods: \[\]/);
  const trigger = JSON.parse(readFileSync(new URL('../recharge-stats.trigger.json', import.meta.url), 'utf8'));
  assert.equal(trigger.target, 'rechargeStats');
  assert.equal(trigger.cron, '0 3 * * *');
});

test('ignores legacy embedded amounts and preserves unrelated user fields', async () => {
  for (const version of [1, 2, 3]) {
    reset();
    state.users[0].rechargeStats = { version, asOf: NOW - 1, amount15d: 1, amount30d: 1, amountTotal: 1 };
    const originalUser = structuredClone(state.users[0]);
    state.order = [order('paid', DAY, { goodsPrice: 101, itemCount: 3 })];
    await run();
    const stats = statsFor();
    assert.equal(stats.version, 8);
    assert.equal(stats.unit, 'fen');
    assert.equal(stats.asOf, CUTOFF);
    assert.equal(stats.amount15d, 303);
    assert.equal(stats.amount30d, 303);
    assert.equal(stats.amountTotal, 303);
    assert.deepEqual(withoutVip(state.users[0]), originalUser);
  }
});

test('payments between midnight and 03:00 are included on the following daily run', async () => {
  reset();
  state.order = [order('midnight', -1), order('02:00', -2 * 3600000 - 1)];
  await run();
  assert.equal(statsFor().amountTotal, 0);
  try {
    Date.now = () => NOW + DAY;
    await run();
    assert.equal(statsFor().amount15d, 200);
    assert.equal(statsFor().amount30d, 200);
    assert.equal(statsFor().amountTotal, 200);
  } finally { Date.now = () => NOW; }
});

test('midnight, scheduled, delayed and manual runs on the same Beijing date have the same cutoff', async () => {
  reset();
  state.order = [order('yesterday-end', 0), order('today-start', -1)];
  try {
    for (const time of ['00:00:00.000', '01:00:00.000', '03:00:00.000', '03:27:16.321', '12:00:00.000', '23:59:59.999']) {
      Date.now = () => Date.parse(`2026-09-15T${time}+08:00`);
      await run();
      const stats = statsFor();
      assert.equal(stats.asOf, CUTOFF, time);
      assert.equal(stats.amount15d, 100, time);
      assert.equal(stats.amount30d, 100, time);
      assert.equal(stats.amountTotal, 100, time);
    }
  } finally { Date.now = () => NOW; }
});

test('Beijing midnight advances the cutoff across month, year and leap-day boundaries', async () => {
  try {
    for (const [runAt, cutoff] of [
      ['2026-10-01T00:00:00+08:00', '2026-09-30T23:59:59.999+08:00'],
      ['2027-01-01T03:00:00+08:00', '2026-12-31T23:59:59.999+08:00'],
      ['2028-03-01T12:00:00+08:00', '2028-02-29T23:59:59.999+08:00'],
    ]) {
      reset();
      Date.now = () => Date.parse(runAt);
      await run();
      assert.equal(statsFor().asOf, Date.parse(cutoff));
    }
  } finally { Date.now = () => NOW; }
});

test('writes VIP0 for a confirmed never-payer and null for a paid flag with missing history', async () => {
  reset();
  state.users.push({ _id: 'free', openid: 'free', pay_user: false, register_time: CUTOFF - 10 * DAY });
  await run();
  assert.equal(userFor('free').vip_level, 0);
  assert.equal(statsFor('free').data_valid, true);
  assert.equal(statsFor().paid_status, 'unknown');
  assert.equal(userFor().vip_level, null);
  assert.equal(statsFor().data_valid, false);
  assert.ok(statsFor().invalid_reasons.includes('paid_history_missing_or_after_cutoff'));
});

test('daily maintenance writes VIP and factors and keeps the entire last good tuple on missing history', async () => {
  reset();
  state.users[0].register_time = CUTOFF - 90 * DAY;
  state.order = Array.from({ length: 30 }, (_, i) => order(`pay-${i}`, i * DAY, { goodsPrice: 600, itemid: 'coin_pack' }));
  await run();
  const before = structuredClone(statsFor());
  assert.equal(userFor().vip_level, 4);
  assert.equal(before.rfm.pay_days_30d, 30);
  assert.equal(before.ticket_factors.general.anchor_fen, 600);
  assert.equal(before.profile_calculated_as_of, CUTOFF + 1);
  state.order = [];
  try {
    Date.now = () => NOW + DAY;
    await run();
  } finally { Date.now = () => NOW; }
  const held = statsFor();
  assert.equal(held.data_valid, false);
  for (const key of ['rfm', 'spending', 'ticket_factors', 'profile_rule_version', 'profile_calculated_as_of']) {
    assert.deepEqual(held[key], before[key]);
  }
  assert.equal(held.asOf, CUTOFF + DAY);
  assert.equal(held.has_ever_paid, true);
  state.users[0].pay_user = false;
  Date.now = () => NOW + 2 * DAY;
  try { await run(); } finally { Date.now = () => NOW; }
  assert.equal(userFor().vip_level, 4);
  assert.equal(statsFor().has_ever_paid, true);
  assert.equal(statsFor().paid_status, 'unknown');
});

test('invalid calculation preserves a good tuple committed during the calculation, not an earlier read', async () => {
  reset();
  const lastGood = {
    _id: 'u1', asOf: CUTOFF - DAY, updatedAt: NOW - 1, has_ever_paid: true,
    vip_level: 3, profile_calculated_as_of: CUTOFF - DAY + 1,
    profile_rule_version: 'earlier-rule', rfm: { score: 3, last_effective_paid_at: CUTOFF - DAY }, spending: { daily_reference_fen: 1000 },
    ticket_factors: { general: { anchor_fen: 3000 } },
  };
  state.beforeWrite = () => { state.userRechargeStats = [structuredClone(lastGood)]; };
  await run();
  assert.equal(userFor().vip_level, 3);
  assert.deepEqual(statsFor().ticket_factors, lastGood.ticket_factors);
  assert.equal(statsFor().data_valid, false);
  assert.equal(statsFor().has_ever_paid, true);
});

test('a paid identity saved during the batch read cannot be overwritten with VIP0', async () => {
  reset();
  state.users[0].pay_user = false;
  state.beforeWrite = () => { state.userRechargeStats = [{
    _id: 'u1', asOf: CUTOFF - DAY, updatedAt: NOW - 1, has_ever_paid: true, vip_level: 2,
    rfm: { score: 2, last_effective_paid_at: CUTOFF - DAY }, profile_calculated_as_of: CUTOFF - DAY + 1,
  }]; };
  await run();
  assert.equal(userFor().vip_level, 2);
  assert.equal(statsFor().has_ever_paid, true);
  assert.equal(statsFor().data_valid, false);
  assert.equal(statsFor().paid_status, 'unknown');
});

test('conflicting duplicates hold the old profile; corrected records restore valid calculation', async () => {
  reset();
  state.order = [order('a', DAY, { goodsPrice: 6800 }), order('b', DAY, { outTradeNo: 'a', goodsPrice: 300 })];
  await run();
  assert.equal(statsFor().duplicateConflictCount, 1);
  assert.equal(userFor().vip_level, null);
  state.order.pop();
  await run();
  assert.equal(statsFor().data_valid, true);
  assert.equal(userFor().vip_level, 2);
});

test('UTC payment confirmation is not shifted and missing SKU is visible in quality metadata', async () => {
  reset();
  state.order = [order('paid', DAY, { chargeTime: 0, time: 0,
    goldMiner: { paidAt: CUTOFF - 3600000 } })];
  await run();
  assert.equal(statsFor().rfm.last_effective_paid_at, CUTOFF - 3600000);
  assert.equal(statsFor().fallbackTimeOrderCount, 0);
  assert.ok(statsFor().quality_warnings.includes('missing_sku_uses_normal_weight'));
});

test('verified never-payers initialize once and skip order queries and writes on later days', async () => {
  reset();
  state.users[0].pay_user = false;
  const first = await run();
  assert.equal(first.data.processedUsers, 1);
  assert.equal(userFor().vip_level, 0);
  assert.equal(statsFor().data_valid, true);
  assert.equal(state.orderQueries.length, 1, 'initialization verifies the order source');
  const initialized = structuredClone(statsFor());
  state.reads = []; state.orderQueries = []; state.writes = 0;
  try {
    Date.now = () => NOW + DAY;
    const second = await run();
    assert.equal(second.data.scannedUsers, 1);
    assert.equal(second.data.skippedNeverPaidUsers, 1);
    assert.equal(second.data.processedUsers, 0);
    assert.equal(second.data.updatedUsers, 0);
    assert.equal(state.orderQueries.length, 0);
    assert.equal(state.writes, 0);
    assert.deepEqual(statsFor(), initialized, 'old initialization time remains truthful');
  } finally { Date.now = () => NOW; }
});

test('pages of stable VIP0 identities are skipped without starving the following paying users', async () => {
  reset();
  state.users = Array.from({ length: 205 }, (_, i) => ({
    _id: `free-${String(i).padStart(3, '0')}`, openid: `free-${i}`, pay_user: false,
  }));
  state.users.push({ _id: 'zz-paid', openid: 'paying', pay_user: true });
  state.order = [order('paid', DAY, { openid: 'paying', goodsPrice: 3000 })];
  await run();
  state.orderQueries = []; state.writes = 0;
  try {
    Date.now = () => NOW + DAY;
    const result = await run();
    assert.equal(result.data.scannedUsers, 206);
    assert.equal(result.data.skippedNeverPaidUsers, 205);
    assert.equal(result.data.processedUsers, 1);
    assert.equal(state.writes, 1);
    assert.equal(state.orderQueries.length, 1);
    assert.deepEqual(state.orderQueries[0].openid.$in, ['paying']);
    assert.equal(statsFor('zz-paid').asOf, CUTOFF + DAY);
  } finally { Date.now = () => NOW; }
});

test('new payments re-enroll VIP0 users automatically and keep aging paid users in maintenance', async () => {
  reset(); state.users[0].pay_user = false;
  await run();
  state.users[0].pay_user = true;
  state.order = [order('first-pay', DAY, { goodsPrice: 6800 })];
  const result = await run();
  assert.equal(result.data.skippedNeverPaidUsers, 0);
  assert.equal(userFor().vip_level, 2);
  assert.equal(statsFor().has_ever_paid, true);
  try {
    Date.now = () => NOW + 40 * DAY;
    const aged = await run();
    assert.equal(aged.data.processedUsers, 1);
    assert.equal(userFor().vip_level, 1);
  } finally { Date.now = () => NOW; }
});

test('a changed account binding is reverified and does not inherit cached VIP0', async () => {
  reset(); state.users[0].pay_user = false;
  await run();
  state.users[0].openid = 'migrated';
  state.order = [order('prior-payment', DAY, { openid: 'migrated' })];
  state.orderQueries = [];
  await run();
  assert.deepEqual(state.orderQueries[0].openid.$in, ['migrated']);
  assert.equal(statsFor().data_valid, false);
  assert.equal(statsFor().paid_status, 'unknown');
  assert.ok(statsFor().invalid_reasons.includes('paid_status_conflict'));
  state.users[0].pay_user = true;
  await run();
  assert.equal(statsFor().data_valid, true);
  assert.equal(userFor().vip_level, 1);
});

test('registration changes, rule changes and old schemas trigger re-initialization', async () => {
  for (const change of ['registration', 'rule', 'schema']) {
    reset(); state.users[0].pay_user = false;
    await run();
    if (change === 'registration') state.users[0].register_time = CUTOFF - 30 * DAY;
    if (change === 'rule') statsFor().profile_rule_version = 'old-rule';
    if (change === 'schema') statsFor().version = 6;
    state.orderQueries = [];
    const result = await run();
    assert.equal(result.data.processedUsers, 1, change);
    assert.equal(state.orderQueries.length, 1, change);
    assert.equal(userFor().vip_level, 0);
    assert.equal(statsFor().version, 8);
  }
});

test('deleted or unknown profiles require identity and order verification, not a default VIP0', async () => {
  reset(); state.users[0].pay_user = false;
  await run();
  state.userRechargeStats = [];
  state.order = [order('unmatched-success', DAY)];
  await run();
  assert.equal(userFor().vip_level, 0);
  assert.equal(userFor().vip_profile.data_valid, false);
  assert.equal(statsFor().paid_status, 'unknown');
  assert.equal(statsFor().data_valid, false);
  // Still unknown on subsequent runs: never cache invalid profiles as free users.
  state.orderQueries = [];
  const result = await run();
  assert.equal(result.data.processedUsers, 1);
  assert.equal(state.orderQueries.length, 1);
});

test('a new payment after the daily cutoff invalidates cached VIP0 until next daily calculation', async () => {
  reset(); state.users[0].pay_user = false;
  await run();
  state.users[0].pay_user = true;
  state.order = [order('today-first', -1)];
  await run();
  assert.equal(statsFor().data_valid, false);
  assert.equal(statsFor().has_ever_paid, true);
  assert.equal(statsFor().paid_status, 'paid');
  try { Date.now = () => NOW + DAY; await run(); } finally { Date.now = () => NOW; }
  assert.equal(statsFor().data_valid, true);
  assert.equal(userFor().vip_level, 1);
});


test('VIP is published only on users with matching computation and identity metadata', async () => {
  reset();
  state.users[0].register_time = new Date(CUTOFF - 40 * DAY);
  state.users[0].coins = 123;
  state.order = [order('paid', DAY, { goodsPrice: 6800 })];
  const result = await run();
  assert.equal(userFor().vip_level, 2);
  assert.equal(userFor().coins, 123);
  assert.equal(userFor().vip_profile.data_valid, true);
  assert.equal(userFor().vip_profile.as_of, statsFor().asOf);
  assert.equal(userFor().vip_profile.updated_at, statsFor().updatedAt);
  assert.equal(userFor().vip_profile.calculated_as_of, statsFor().profile_calculated_as_of);
  assert.equal(userFor().vip_profile.identity.register_time, CUTOFF - 40 * DAY);
  assert.equal(result.data.updatedVipUsers, 1);
  assert.ok(!Object.hasOwn(statsFor(), 'vip_level'));
  assert.equal((await run()).data.updatedVipUsers, 0);
});

test('version 7 migrates held RFM without reading the old VIP field, including after it is deleted', async () => {
  for (const oldField of [{ vip_level: 5 }, {}]) {
    reset();
    state.userRechargeStats = [{
      _id: 'u1', version: 7, asOf: CUTOFF - DAY, updatedAt: NOW - DAY, has_ever_paid: true,
      profile_calculated_as_of: CUTOFF - DAY + 1, profile_rule_version: 'previous-rule',
      rfm: { score: 3.6, last_effective_paid_at: CUTOFF - DAY }, ...oldField,
    }];
    await run();
    assert.equal(userFor().vip_level, 4);
    assert.equal(userFor().vip_profile.data_valid, false);
    assert.equal(userFor().vip_profile.calculated_as_of, CUTOFF - DAY + 1);
    assert.equal(userFor().vip_profile.rule_version, 'previous-rule');
    assert.ok(!Object.hasOwn(statsFor(), 'vip_level'));
  }
});

test('partial publication failure propagates and rerun repairs users before VIP0 can be skipped', async () => {
  reset(); userFor().pay_user = false;
  state.failUserWrite = true;
  await assert.rejects(run(), /users write failed/);
  assert.equal(statsFor().version, 8);
  assert.equal(userFor().vip_level, undefined);
  state.failUserWrite = false;
  const result = await run();
  assert.equal(result.data.processedUsers, 1);
  assert.equal(result.data.updatedUsers, 0);
  assert.equal(result.data.updatedVipUsers, 1);
  assert.equal(userFor().vip_level, 0);
  assert.equal(userFor().vip_profile.data_valid, true);
  assert.equal((await run()).data.skippedNeverPaidUsers, 1);
});

test('a newer users snapshot survives an older publication', async () => {
  reset(); state.order = [order('paid', DAY)];
  const newer = { as_of: CUTOFF + DAY, updated_at: NOW + DAY, data_valid: true,
    calculated_as_of: CUTOFF + DAY + 1, rule_version: 'newer-rule', has_ever_paid: true };
  state.beforeUserWrite = () => { userFor().vip_level = 5; userFor().vip_profile = structuredClone(newer); };
  await run();
  assert.equal(userFor().vip_level, 5);
  assert.deepEqual(userFor().vip_profile, newer);
});

test('invalid input preserves the latest users grade and its actual calculation time atomically', async () => {
  reset();
  state.beforeUserWrite = () => {
    userFor().vip_level = 3;
    userFor().vip_profile = { as_of: CUTOFF - DAY, updated_at: NOW - DAY,
      calculated_as_of: CUTOFF - DAY + 1, rule_version: 'old-rule', has_ever_paid: true };
  };
  await run();
  assert.equal(userFor().vip_level, 3);
  assert.equal(userFor().vip_profile.data_valid, false);
  assert.equal(userFor().vip_profile.calculated_as_of, CUTOFF - DAY + 1);
  assert.equal(userFor().vip_profile.rule_version, 'old-rule');
});

test('concurrent identity changes invalidate VIP publication and preserve unrelated user writes', async () => {
  for (const change of ['payment', 'openid', 'registration']) {
    reset(); userFor().pay_user = false;
    state.beforeUserWrite = () => {
      if (change === 'payment') userFor().pay_user = true;
      if (change === 'openid') userFor().openid = 'changed';
      if (change === 'registration') userFor().register_time = CUTOFF - DAY;
      userFor().coins = 777;
    };
    await run();
    assert.equal(userFor().vip_profile.data_valid, false, change);
    assert.equal(userFor().coins, 777);
  }
});

test('user deletion between reads and publication does not recreate the account', async () => {
  reset(); state.order = [order('paid', DAY)];
  state.beforeUserWrite = () => { state.users = []; };
  await run();
  assert.equal(state.users.length, 0);
});

test('paid evidence on users prevents VIP0 after a stats deletion and pay flag rollback', async () => {
  reset(); state.order = [order('paid', DAY, { goodsPrice: 6800 })];
  await run();
  state.userRechargeStats = []; state.order = []; userFor().pay_user = false;
  await run();
  assert.equal(userFor().vip_level, 2);
  assert.equal(userFor().vip_profile.data_valid, false);
  assert.equal(statsFor().has_ever_paid, true);
  assert.equal(statsFor().paid_status, 'unknown');
});

test('publication uses the committed winner rather than the attempted calculation', async () => {
  reset(); state.order = [order('paid', DAY, { goodsPrice: 6800 })];
  await run();
  const winner = structuredClone(statsFor());
  winner.updatedAt = NOW + 60000;
  winner.rfm.score = 3.6;
  state.beforeWrite = () => { state.userRechargeStats = [structuredClone(winner)]; };
  await run();
  assert.equal(userFor().vip_level, 4);
  assert.equal(userFor().vip_profile.updated_at, NOW + 60000);
  assert.equal(userFor().vip_profile.data_valid, true);
});