memind/memind-canary-proxy.test.mjs

import assert from 'node:assert/strict';
import http from 'node:http';
import test from 'node:test';

import { createCanaryPolicy } from './release-gate/canary-routing.mjs';
import {
  createCanaryIdentityResolver,
  createMemindCanaryProxy,
  parseCookieHeader,
  parseWechatOpenid,
  verifyCanaryPolicySelectors,
} from './memind-canary-proxy.mjs';

function listen(server) {
  return new Promise((resolve, reject) => {
    server.once('error', reject);
    server.listen(0, '127.0.0.1', () => {
      const address = server.address();
      resolve(`http://127.0.0.1:${address.port}`);
    });
  });
}

function close(server) {
  return new Promise((resolve) => server.close(resolve));
}

function request(baseUrl, {
  method = 'GET',
  path = '/',
  headers = {},
  body = '',
} = {}) {
  const target = new URL(path, baseUrl);
  return new Promise((resolve, reject) => {
    const req = http.request(
      {
        hostname: target.hostname,
        port: target.port,
        path: `${target.pathname}${target.search}`,
        method,
        headers: {
          ...headers,
          ...(body ? { 'content-length': Buffer.byteLength(body) } : {}),
        },
      },
      (res) => {
        const chunks = [];
        res.on('data', (chunk) => chunks.push(chunk));
        res.on('end', () => resolve({
          status: res.statusCode,
          headers: res.headers,
          body: Buffer.concat(chunks).toString('utf8'),
        }));
      },
    );
    req.once('error', reject);
    req.end(body);
  });
}

function upstream(name) {
  return http.createServer((req, res) => {
    if (req.url === '/api/status') {
      res.writeHead(200, { 'x-memind-runtime-role': name });
      res.end('ok');
      return;
    }
    const chunks = [];
    req.on('data', (chunk) => chunks.push(chunk));
    req.on('end', () => {
      res.writeHead(200, { 'content-type': 'text/plain' });
      res.end(`${name}:${Buffer.concat(chunks).toString('utf8')}`);
    });
  });
}

function candidateThatStopsAfterHealth() {
  const server = http.createServer((req, res) => {
    if (req.url !== '/api/status') {
      res.writeHead(200);
      res.end('candidate');
      return;
    }
    server.close();
    res.writeHead(200, { 'x-memind-runtime-role': 'candidate' });
    res.end('ok');
  });
  return server;
}

test('cookie and WeChat parsers preserve immutable routing inputs', () => {
  const cookies = parseCookieHeader('a=1; tkmind_user_session=abc%2D123');
  assert.equal(cookies.get('tkmind_user_session'), 'abc-123');
  assert.equal(
    parseWechatOpenid(
      '<xml><FromUserName><![CDATA[openid-target]]></FromUserName></xml>',
    ),
    'openid-target',
  );
  assert.equal(parseWechatOpenid('<xml></xml>'), '');
});

test('identity resolver hashes sessions and maps WeChat openid to bound user id', async () => {
  const calls = [];
  const resolver = createCanaryIdentityResolver({
    pool: {
      async query(sql, params) {
        calls.push({ sql, params });
        if (sql.includes('h5_login_sessions')) {
          return [[{ user_id: 'user-john', username: 'john' }]];
        }
        return [[{ user_id: 'wx_ul610et8', username: 'wechat_user' }]];
      },
    },
    appId: 'app-1',
    now: () => 123,
  });

  assert.deepEqual(await resolver.fromSessionToken('token-1'), {
    id: 'user-john',
    username: 'john',
  });
  assert.match(calls[0].params[0], /^[a-f0-9]{64}$/);
  assert.equal(calls[0].params[1], 123);

  assert.deepEqual(await resolver.fromWechatOpenid('openid-1'), {
    id: 'wx_ul610et8',
    username: 'wechat_user',
    wechatUserId: 'wechat_user',
  });
  assert.deepEqual(calls[1].params, ['app-1', 'openid-1']);
});

test('WeChat canary selector validates the bound unique username, not a production UUID', async () => {
  const calls = [];
  const summary = await verifyCanaryPolicySelectors(
    {
      async query(sql, params) {
        calls.push({ sql, params });
        if (sql.includes('LOWER(username)')) return [[{ id: 'john-uuid' }]];
        return [[{ id: 'wechat-uuid', has_wechat: 1 }]];
      },
    },
    createCanaryPolicy({
      usernames: ['john'],
      wechatUserIds: ['wx_ul610et8'],
    }),
  );

  assert.deepEqual(summary, { userIds: 0, usernames: 1, wechatUserIds: 1 });
  assert.match(calls[1].sql, /WHERE u\.username = \?/);
  assert.deepEqual(calls[1].params, ['wx_ul610et8']);
});

test('proxy routes only immutable canary identities and preserves WeChat bodies', async (t) => {
  const stableServer = upstream('stable');
  const candidateServer = upstream('candidate');
  const stableUrl = await listen(stableServer);
  const candidateUrl = await listen(candidateServer);
  const policy = createCanaryPolicy({
    usernames: ['john'],
    wechatUserIds: ['wx_ul610et8'],
  });
  const identityResolver = {
    async fromSessionToken(token) {
      if (token === 'john-token') return { id: 'user-john', username: 'john' };
      if (token === 'same-name-token') {
        return { id: 'other-user', username: 'other', displayName: '唐' };
      }
      return null;
    },
    async fromWechatOpenid(openid) {
      if (openid === 'openid-target') {
        return {
          id: 'wechat-user-uuid',
          username: 'wx_ul610et8',
          wechatUserId: 'wx_ul610et8',
        };
      }
      return null;
    },
  };
  const proxy = createMemindCanaryProxy({
    stableTarget: stableUrl,
    candidateTarget: candidateUrl,
    policy,
    identityResolver,
    diagnosticSecret: 'diagnostic-secret',
    candidateHealthTtlMs: 0,
  });
  const proxyUrl = await listen(proxy.server);
  t.after(async () => {
    await close(proxy.server);
    await close(stableServer);
    if (candidateServer.listening) await close(candidateServer);
  });

  const john = await request(proxyUrl, {
    headers: { cookie: 'tkmind_user_session=john-token' },
  });
  assert.equal(john.body, 'candidate:');
  assert.equal(john.headers['x-memind-runtime-route'], 'candidate');

  const missingIdentity = await request(proxyUrl);
  assert.equal(missingIdentity.body, 'stable:');
  assert.equal(missingIdentity.headers['x-memind-runtime-route'], 'stable');

  const sameNickname = await request(proxyUrl, {
    headers: { cookie: 'tkmind_user_session=same-name-token' },
  });
  assert.equal(sameNickname.body, 'stable:');

  const xml =
    '<xml><FromUserName><![CDATA[openid-target]]></FromUserName>'
    + '<Content><![CDATA[hello]]></Content></xml>';
  const wechat = await request(proxyUrl, {
    method: 'POST',
    path: '/webhooks/wechat-mp/messages',
    headers: { 'content-type': 'text/xml' },
    body: xml,
  });
  assert.equal(wechat.body, `candidate:${xml}`);
  assert.equal(wechat.headers['x-memind-runtime-route'], 'candidate');

  const health = await request(proxyUrl, {
    path: '/__memind_canary/health',
    headers: { 'x-memind-canary-secret': 'diagnostic-secret' },
  });
  assert.equal(health.status, 200);
  assert.deepEqual(JSON.parse(health.body), {
    ok: true,
    stable: true,
    candidate: true,
    policy: { userIds: 0, usernames: 1, wechatUserIds: 1 },
  });

  await close(candidateServer);
  const routeProbe = await request(proxyUrl, {
    path: '/__memind_canary/route-probe?username=john',
    headers: { 'x-memind-canary-secret': 'diagnostic-secret' },
  });
  assert.deepEqual(JSON.parse(routeProbe.body), {
    route: 'stable',
    configuredTarget: 'candidate',
    candidate: false,
  });
  const fallback = await request(proxyUrl, {
    headers: { cookie: 'tkmind_user_session=john-token' },
  });
  assert.equal(fallback.body, 'stable:');
  assert.equal(fallback.headers['x-memind-runtime-route'], 'stable');
});

test('identity lookup errors fail closed to stable', async (t) => {
  const stableServer = upstream('stable');
  const candidateServer = upstream('candidate');
  const stableUrl = await listen(stableServer);
  const candidateUrl = await listen(candidateServer);
  const proxy = createMemindCanaryProxy({
    stableTarget: stableUrl,
    candidateTarget: candidateUrl,
    policy: createCanaryPolicy({ usernames: ['john'] }),
    identityResolver: {
      async fromSessionToken() {
        throw new Error('database unavailable');
      },
      async fromWechatOpenid() {
        throw new Error('database unavailable');
      },
    },
    diagnosticSecret: 'diagnostic-secret',
    candidateHealthTtlMs: 0,
    logger: { warn() {} },
  });
  const proxyUrl = await listen(proxy.server);
  t.after(async () => {
    await close(proxy.server);
    await close(stableServer);
    await close(candidateServer);
  });

  const response = await request(proxyUrl, {
    headers: { cookie: 'tkmind_user_session=john-token' },
  });
  assert.equal(response.body, 'stable:');
  assert.equal(response.headers['x-memind-runtime-route'], 'stable');
});

test('candidate health fails closed when its DeepSeek compatibility dependency is down', async (t) => {
  const stableServer = upstream('stable');
  const candidateServer = upstream('candidate');
  let dependencyHealthy = true;
  const dependencyServer = http.createServer((req, res) => {
    res.writeHead(req.url === '/health' && dependencyHealthy ? 200 : 503);
    res.end();
  });
  const stableUrl = await listen(stableServer);
  const candidateUrl = await listen(candidateServer);
  const dependencyUrl = await listen(dependencyServer);
  const proxy = createMemindCanaryProxy({
    stableTarget: stableUrl,
    candidateTarget: candidateUrl,
    candidateHealthDependencies: [`${dependencyUrl}/health`],
    policy: createCanaryPolicy({ usernames: ['john'] }),
    identityResolver: {
      async fromSessionToken() {
        return { id: 'user-john', username: 'john' };
      },
      async fromWechatOpenid() {
        return null;
      },
    },
    diagnosticSecret: 'diagnostic-secret',
    candidateHealthTtlMs: 0,
  });
  const proxyUrl = await listen(proxy.server);
  t.after(async () => {
    await close(proxy.server);
    await close(stableServer);
    await close(candidateServer);
    await close(dependencyServer);
  });

  const candidate = await request(proxyUrl, {
    headers: { cookie: 'tkmind_user_session=john-token' },
  });
  assert.equal(candidate.headers['x-memind-runtime-route'], 'candidate');

  dependencyHealthy = false;
  const fallback = await request(proxyUrl, {
    headers: { cookie: 'tkmind_user_session=john-token' },
  });
  assert.equal(fallback.body, 'stable:');
  assert.equal(fallback.headers['x-memind-runtime-route'], 'stable');
});

test('candidate health requires the explicit candidate runtime role', async (t) => {
  const stableServer = upstream('stable');
  const wrongRoleServer = upstream('stable');
  const stableUrl = await listen(stableServer);
  const wrongRoleUrl = await listen(wrongRoleServer);
  const proxy = createMemindCanaryProxy({
    stableTarget: stableUrl,
    candidateTarget: wrongRoleUrl,
    policy: createCanaryPolicy({ usernames: ['john'] }),
    identityResolver: {
      async fromSessionToken() {
        return { id: 'user-john', username: 'john' };
      },
      async fromWechatOpenid() {
        return null;
      },
    },
    diagnosticSecret: 'diagnostic-secret',
    candidateHealthTtlMs: 0,
  });
  const proxyUrl = await listen(proxy.server);
  t.after(async () => {
    await close(proxy.server);
    await close(stableServer);
    await close(wrongRoleServer);
  });

  const response = await request(proxyUrl, {
    headers: { cookie: 'tkmind_user_session=john-token' },
  });
  assert.equal(response.body, 'stable:');
  assert.equal(response.headers['x-memind-runtime-route'], 'stable');
});

test('candidate connect race retries only replay-safe requests on stable', async (t) => {
  const stableServer = upstream('stable');
  const candidateServer = candidateThatStopsAfterHealth();
  const stableUrl = await listen(stableServer);
  const candidateUrl = await listen(candidateServer);
  const proxy = createMemindCanaryProxy({
    stableTarget: stableUrl,
    candidateTarget: candidateUrl,
    policy: createCanaryPolicy({ usernames: ['john'] }),
    identityResolver: {
      async fromSessionToken() {
        return { id: 'user-john', username: 'john' };
      },
      async fromWechatOpenid() {
        return null;
      },
    },
    diagnosticSecret: 'diagnostic-secret',
    candidateHealthTtlMs: 0,
  });
  const proxyUrl = await listen(proxy.server);
  t.after(async () => {
    await close(proxy.server);
    await close(stableServer);
    if (candidateServer.listening) await close(candidateServer);
  });

  const response = await request(proxyUrl, {
    headers: { cookie: 'tkmind_user_session=john-token' },
  });
  assert.equal(response.status, 200);
  assert.equal(response.body, 'stable:');
  assert.equal(response.headers['x-memind-runtime-route'], 'stable');
});

test('candidate connect race never replays an unbuffered write to stable', async (t) => {
  const stableServer = upstream('stable');
  const candidateServer = candidateThatStopsAfterHealth();
  const stableUrl = await listen(stableServer);
  const candidateUrl = await listen(candidateServer);
  const proxy = createMemindCanaryProxy({
    stableTarget: stableUrl,
    candidateTarget: candidateUrl,
    policy: createCanaryPolicy({ usernames: ['john'] }),
    identityResolver: {
      async fromSessionToken() {
        return { id: 'user-john', username: 'john' };
      },
      async fromWechatOpenid() {
        return null;
      },
    },
    diagnosticSecret: 'diagnostic-secret',
    candidateHealthTtlMs: 0,
  });
  const proxyUrl = await listen(proxy.server);
  t.after(async () => {
    await close(proxy.server);
    await close(stableServer);
    if (candidateServer.listening) await close(candidateServer);
  });

  const response = await request(proxyUrl, {
    method: 'POST',
    path: '/api/write',
    headers: {
      cookie: 'tkmind_user_session=john-token',
      'content-type': 'application/json',
    },
    body: '{"value":1}',
  });
  assert.equal(response.status, 502);
  assert.match(response.body, /canary upstream unavailable/);
});