feat: add guarded portal canary release

memind-canary-proxy.mjs

@@ -0,0 +1,569 @@
+#!/usr/bin/env node
+import crypto from 'node:crypto';
+import http from 'node:http';
+import { pathToFileURL } from 'node:url';
+
+import mysql from 'mysql2/promise';
+
+import {
+  canaryPolicyFromEnv,
+  resolveCanaryTarget,
+} from './release-gate/canary-routing.mjs';
+import { loadMemindEnvFiles } from './scripts/memind-runtime-profile.mjs';
+
+const SESSION_COOKIE = 'tkmind_user_session';
+const DEFAULT_BODY_LIMIT = 2 * 1024 * 1024;
+const RETRYABLE_CONNECT_ERRORS = new Set([
+  'ECONNREFUSED',
+  'EHOSTUNREACH',
+  'ENETUNREACH',
+]);
+
+function csvValues(value) {
+  return String(value ?? '')
+    .split(',')
+    .map((item) => item.trim())
+    .filter(Boolean);
+}
+
+export function parseCookieHeader(header) {
+  const cookies = new Map();
+  for (const item of String(header ?? '').split(';')) {
+    const separator = item.indexOf('=');
+    if (separator < 1) continue;
+    const name = item.slice(0, separator).trim();
+    const rawValue = item.slice(separator + 1).trim();
+    if (!name) continue;
+    try {
+      cookies.set(name, decodeURIComponent(rawValue));
+    } catch {
+      cookies.set(name, rawValue);
+    }
+  }
+  return cookies;
+}
+
+export function parseWechatOpenid(xml) {
+  const match = String(xml ?? '').match(
+    /<FromUserName><!\[CDATA\[([\s\S]*?)\]\]><\/FromUserName>/i,
+  );
+  return String(match?.[1] ?? '').trim();
+}
+
+function sessionTokenHash(token) {
+  return crypto.createHash('sha256').update(token).digest('hex');
+}
+
+function normalizeLoopbackTarget(value, label) {
+  const target = new URL(String(value ?? ''));
+  if (
+    target.protocol !== 'http:'
+    || !['127.0.0.1', 'localhost', '::1'].includes(target.hostname)
+  ) {
+    throw new Error(`${label} must be a loopback http URL`);
+  }
+  target.pathname = target.pathname.replace(/\/+$/, '');
+  target.search = '';
+  target.hash = '';
+  return target;
+}
+
+function createIdentityPool(env = process.env) {
+  const poolOptions = {
+    waitForConnections: true,
+    connectionLimit: Math.max(
+      1,
+      Math.min(8, Number(env.MEMIND_CANARY_DB_POOL_SIZE ?? 4)),
+    ),
+    queueLimit: 32,
+  };
+  if (env.DATABASE_URL) {
+    return mysql.createPool({ uri: env.DATABASE_URL, ...poolOptions });
+  }
+  if (!env.MYSQL_HOST || !env.MYSQL_DATABASE) {
+    throw new Error('Canary identity routing requires DATABASE_URL or MYSQL_*');
+  }
+  return mysql.createPool({
+    host: env.MYSQL_HOST,
+    port: Number(env.MYSQL_PORT ?? 3306),
+    user: env.MYSQL_USER ?? 'boot',
+    password: env.MYSQL_PASSWORD ?? '',
+    database: env.MYSQL_DATABASE,
+    ...poolOptions,
+  });
+}
+
+export function createCanaryIdentityResolver({
+  pool,
+  appId = '',
+  now = () => Date.now(),
+} = {}) {
+  if (!pool?.query) throw new Error('identity resolver requires a database pool');
+
+  return {
+    async fromSessionToken(token) {
+      if (!token) return null;
+      const [rows] = await pool.query(
+        `SELECT s.user_id, u.username
+         FROM h5_login_sessions s
+         JOIN h5_users u ON u.id = s.user_id
+         WHERE s.token_hash = ?
+           AND s.revoked_at IS NULL
+           AND s.expires_at > ?
+           AND u.status <> 'disabled'
+         LIMIT 1`,
+        [sessionTokenHash(token), now()],
+      );
+      const row = rows[0];
+      return row
+        ? {
+            id: String(row.user_id ?? ''),
+            username: String(row.username ?? ''),
+          }
+        : null;
+    },
+
+    async fromWechatOpenid(openid) {
+      if (!openid || !appId) return null;
+      const [rows] = await pool.query(
+        `SELECT wi.user_id, u.username
+         FROM h5_user_wechat_identities wi
+         JOIN h5_users u ON u.id = wi.user_id
+         WHERE wi.app_id = ?
+           AND wi.openid = ?
+           AND u.status <> 'disabled'
+         LIMIT 1`,
+        [appId, openid],
+      );
+      const row = rows[0];
+      return row
+        ? {
+            id: String(row.user_id ?? ''),
+            username: String(row.username ?? ''),
+            wechatUserId: String(row.username ?? ''),
+          }
+        : null;
+    },
+  };
+}
+
+export async function verifyCanaryPolicySelectors(pool, policy) {
+  const selectors = [
+    ...[...(policy.userIds ?? [])].map((value) => ['id', value]),
+    ...[...(policy.usernames ?? [])].map((value) => ['username', value]),
+    ...[...(policy.wechatUserIds ?? [])].map((value) => ['wechat', value]),
+  ];
+  for (const [kind, value] of selectors) {
+    if (kind === 'username') {
+      const [rows] = await pool.query(
+        `SELECT id FROM h5_users
+         WHERE LOWER(username) = LOWER(?) AND status <> 'disabled'
+         LIMIT 2`,
+        [value],
+      );
+      if (rows.length !== 1) {
+        throw new Error('configured canary username does not resolve uniquely');
+      }
+      continue;
+    }
+    const lookupColumn = kind === 'wechat' ? 'u.username' : 'u.id';
+    const [rows] = await pool.query(
+      `SELECT u.id,
+              EXISTS(
+                SELECT 1 FROM h5_user_wechat_identities wi WHERE wi.user_id = u.id
+              ) AS has_wechat
+       FROM h5_users u
+       WHERE ${lookupColumn} = ? AND u.status <> 'disabled'
+       LIMIT 2`,
+      [value],
+    );
+    if (rows.length !== 1) {
+      throw new Error('configured canary immutable user selector does not resolve uniquely');
+    }
+    if (kind === 'wechat' && Number(rows[0].has_wechat ?? 0) !== 1) {
+      throw new Error('configured WeChat canary user has no bound immutable identity');
+    }
+  }
+  return {
+    userIds: policy.userIds?.size ?? 0,
+    usernames: policy.usernames?.size ?? 0,
+    wechatUserIds: policy.wechatUserIds?.size ?? 0,
+  };
+}
+
+function requestBody(req, limit = DEFAULT_BODY_LIMIT) {
+  return new Promise((resolve, reject) => {
+    const chunks = [];
+    let bytes = 0;
+    req.on('data', (chunk) => {
+      bytes += chunk.length;
+      if (bytes > limit) {
+        reject(Object.assign(new Error('request body too large'), { statusCode: 413 }));
+        req.destroy();
+        return;
+      }
+      chunks.push(chunk);
+    });
+    req.on('end', () => resolve(Buffer.concat(chunks)));
+    req.on('error', reject);
+  });
+}
+
+function healthRequest(target, timeoutMs, expectedRole = '') {
+  return new Promise((resolve) => {
+    const request = http.request(
+      {
+        hostname: target.hostname,
+        port: target.port,
+        method: 'GET',
+        path: `${target.pathname}/api/status`.replace(/\/{2,}/g, '/'),
+        headers: { connection: 'close' },
+      },
+      (response) => {
+        response.resume();
+        response.once('end', () => {
+          const runtimeRole = String(
+            response.headers['x-memind-runtime-role'] ?? '',
+          ).trim().toLowerCase();
+          resolve(
+            response.statusCode === 200
+            && (!expectedRole || runtimeRole === expectedRole),
+          );
+        });
+      },
+    );
+    request.setTimeout(timeoutMs, () => request.destroy());
+    request.once('error', () => resolve(false));
+    request.end();
+  });
+}
+
+function appendForwardedFor(current, remoteAddress) {
+  const values = [String(current ?? '').trim(), String(remoteAddress ?? '').trim()]
+    .filter(Boolean);
+  return values.join(', ');
+}
+
+export function createMemindCanaryProxy({
+  stableTarget,
+  candidateTarget,
+  policy,
+  identityResolver,
+  diagnosticSecret,
+  candidateHealthTtlMs = 1_000,
+  candidateHealthTimeoutMs = 750,
+  logger = console,
+} = {}) {
+  const stable = normalizeLoopbackTarget(stableTarget, 'stable target');
+  const candidate = normalizeLoopbackTarget(candidateTarget, 'candidate target');
+  if (!policy) throw new Error('canary proxy requires a routing policy');
+  if (!identityResolver) throw new Error('canary proxy requires an identity resolver');
+  if (!diagnosticSecret) throw new Error('canary proxy requires a diagnostic secret');
+
+  let healthCache = { checkedAt: 0, healthy: false };
+  let healthPromise = null;
+
+  const candidateHealthy = async ({ force = false } = {}) => {
+    const currentTime = Date.now();
+    if (!force && currentTime - healthCache.checkedAt < candidateHealthTtlMs) {
+      return healthCache.healthy;
+    }
+    if (!healthPromise) {
+      healthPromise = healthRequest(candidate, candidateHealthTimeoutMs, 'candidate')
+        .then((healthy) => {
+          healthCache = { checkedAt: Date.now(), healthy };
+          return healthy;
+        })
+        .finally(() => {
+          healthPromise = null;
+        });
+    }
+    return healthPromise;
+  };
+
+  const stableHealthy = () => healthRequest(stable, candidateHealthTimeoutMs);
+
+  const resolveIdentity = async (req, bufferedBody) => {
+    if (
+      req.method === 'POST'
+      && String(req.url ?? '').split('?')[0] === '/webhooks/wechat-mp/messages'
+    ) {
+      return identityResolver.fromWechatOpenid(
+        parseWechatOpenid(bufferedBody?.toString('utf8')),
+      );
+    }
+    const token = parseCookieHeader(req.headers.cookie).get(SESSION_COOKIE);
+    return identityResolver.fromSessionToken(token);
+  };
+
+  const selectTarget = async (req, bufferedBody) => {
+    let identity = null;
+    try {
+      identity = await resolveIdentity(req, bufferedBody);
+    } catch (error) {
+      logger.warn?.(
+        'Canary identity resolution failed closed to stable:',
+        error instanceof Error ? error.message : String(error),
+      );
+      return { route: 'stable', target: stable };
+    }
+    if (resolveCanaryTarget(identity, policy) !== 'candidate') {
+      return { route: 'stable', target: stable };
+    }
+    if (!(await candidateHealthy())) {
+      return { route: 'stable', target: stable };
+    }
+    return { route: 'candidate', target: candidate };
+  };
+
+  const proxyRequest = (req, res, selection, bufferedBody, allowFallback = true) => {
+    const headers = {
+      ...req.headers,
+      'x-forwarded-for': appendForwardedFor(
+        req.headers['x-forwarded-for'],
+        req.socket.remoteAddress,
+      ),
+      'x-memind-canary-route': selection.route,
+    };
+    delete headers['x-memind-runtime-route'];
+    if (bufferedBody) {
+      headers['content-length'] = String(bufferedBody.length);
+      delete headers['transfer-encoding'];
+    }
+    const hasDeclaredBody =
+      Boolean(req.headers['content-length'])
+      || Boolean(req.headers['transfer-encoding']);
+    const safeToReplay =
+      Boolean(bufferedBody)
+      || ['GET', 'HEAD', 'OPTIONS'].includes(String(req.method ?? '').toUpperCase())
+      || !hasDeclaredBody;
+
+    let connected = false;
+    const upstream = http.request(
+      {
+        hostname: selection.target.hostname,
+        port: selection.target.port,
+        method: req.method,
+        path: `${selection.target.pathname}${req.url ?? '/'}`.replace(/\/{2,}/g, '/'),
+        headers,
+      },
+      (upstreamResponse) => {
+        const responseHeaders = {
+          ...upstreamResponse.headers,
+          'x-memind-runtime-route': selection.route,
+        };
+        res.writeHead(upstreamResponse.statusCode ?? 502, responseHeaders);
+        upstreamResponse.pipe(res);
+      },
+    );
+    upstream.on('socket', (socket) => {
+      socket.once('connect', () => {
+        connected = true;
+      });
+    });
+    upstream.once('error', (error) => {
+      if (
+        allowFallback
+        && selection.route === 'candidate'
+        && safeToReplay
+        && !connected
+        && RETRYABLE_CONNECT_ERRORS.has(error?.code)
+        && !res.headersSent
+      ) {
+        healthCache = { checkedAt: Date.now(), healthy: false };
+        proxyRequest(
+          req,
+          res,
+          { route: 'stable', target: stable },
+          bufferedBody,
+          false,
+        );
+        return;
+      }
+      if (!res.headersSent) {
+        res.writeHead(502, { 'content-type': 'application/json; charset=utf-8' });
+      }
+      res.end(JSON.stringify({ ok: false, message: 'canary upstream unavailable' }));
+    });
+    if (bufferedBody) {
+      upstream.end(bufferedBody);
+    } else if (!hasDeclaredBody) {
+      upstream.end();
+    } else {
+      req.pipe(upstream);
+    }
+  };
+
+  const server = http.createServer(async (req, res) => {
+    try {
+      const requestUrl = new URL(req.url ?? '/', 'http://127.0.0.1');
+      const pathname = requestUrl.pathname;
+      if (pathname === '/__memind_canary/health') {
+        if (req.headers['x-memind-canary-secret'] !== diagnosticSecret) {
+          res.writeHead(404).end();
+          return;
+        }
+        const [stableOk, candidateOk] = await Promise.all([
+          stableHealthy(),
+          candidateHealthy({ force: true }),
+        ]);
+        res.writeHead(stableOk ? 200 : 503, {
+          'content-type': 'application/json; charset=utf-8',
+          'cache-control': 'no-store',
+        });
+        res.end(JSON.stringify({
+          ok: stableOk,
+          stable: stableOk,
+          candidate: candidateOk,
+          policy: {
+            userIds: policy.userIds?.size ?? 0,
+            usernames: policy.usernames?.size ?? 0,
+            wechatUserIds: policy.wechatUserIds?.size ?? 0,
+          },
+        }));
+        return;
+      }
+      if (pathname === '/__memind_canary/route-probe') {
+        if (req.headers['x-memind-canary-secret'] !== diagnosticSecret) {
+          res.writeHead(404).end();
+          return;
+        }
+        const identity = {
+          id: requestUrl.searchParams.get('id') ?? '',
+          username: requestUrl.searchParams.get('username') ?? '',
+          wechatUserId: requestUrl.searchParams.get('wechat_user_id') ?? '',
+        };
+        const configuredTarget = resolveCanaryTarget(identity, policy);
+        const candidateOk =
+          configuredTarget === 'candidate' ? await candidateHealthy({ force: true }) : false;
+        const route =
+          configuredTarget === 'candidate' && candidateOk ? 'candidate' : 'stable';
+        res.writeHead(200, {
+          'content-type': 'application/json; charset=utf-8',
+          'cache-control': 'no-store',
+        });
+        res.end(JSON.stringify({
+          route,
+          configuredTarget,
+          candidate: candidateOk,
+        }));
+        return;
+      }
+
+      const needsBody =
+        req.method === 'POST'
+        && pathname === '/webhooks/wechat-mp/messages';
+      const bufferedBody = needsBody ? await requestBody(req) : null;
+      const selection = await selectTarget(req, bufferedBody);
+      proxyRequest(req, res, selection, bufferedBody);
+    } catch (error) {
+      const statusCode = Number(error?.statusCode ?? 500);
+      if (!res.headersSent) {
+        res.writeHead(statusCode, { 'content-type': 'application/json; charset=utf-8' });
+      }
+      res.end(JSON.stringify({ ok: false, message: 'canary routing failed closed' }));
+    }
+  });
+
+  server.on('upgrade', async (req, socket, head) => {
+    let selection;
+    try {
+      selection = await selectTarget(req, null);
+    } catch {
+      selection = { route: 'stable', target: stable };
+    }
+    const headers = {
+      ...req.headers,
+      connection: 'Upgrade',
+      upgrade: req.headers.upgrade ?? 'websocket',
+      'x-memind-canary-route': selection.route,
+    };
+    const upstream = http.request({
+      hostname: selection.target.hostname,
+      port: selection.target.port,
+      method: req.method,
+      path: req.url,
+      headers,
+    });
+    upstream.on('upgrade', (response, upstreamSocket, upstreamHead) => {
+      const statusLine =
+        `HTTP/1.1 ${response.statusCode ?? 101} ${response.statusMessage ?? 'Switching Protocols'}\r\n`;
+      const responseHeaders = Object.entries(response.headers)
+        .flatMap(([name, value]) => {
+          const values = Array.isArray(value) ? value : [value];
+          return values.filter((item) => item != null).map((item) => `${name}: ${item}\r\n`);
+        })
+        .join('');
+      socket.write(`${statusLine}${responseHeaders}\r\n`);
+      if (upstreamHead.length) socket.write(upstreamHead);
+      if (head.length) upstreamSocket.write(head);
+      upstreamSocket.pipe(socket);
+      socket.pipe(upstreamSocket);
+    });
+    upstream.once('error', () => socket.destroy());
+    upstream.end();
+  });
+
+  return {
+    server,
+    candidateHealthy,
+    stableHealthy,
+  };
+}
+
+async function main() {
+  const persistentRoot =
+    process.env.MEMIND_PORTAL_H5_ROOT
+    || process.env.MEMIND_CANARY_STABLE_ROOT
+    || process.cwd();
+  loadMemindEnvFiles(persistentRoot);
+  const policy = canaryPolicyFromEnv(process.env);
+  const pool = createIdentityPool(process.env);
+  const selectorSummary = await verifyCanaryPolicySelectors(pool, policy);
+  if (process.argv.includes('--check')) {
+    console.log(JSON.stringify({ ok: true, policy: selectorSummary }));
+    await pool.end();
+    return;
+  }
+
+  const identityResolver = createCanaryIdentityResolver({
+    pool,
+    appId:
+      process.env.H5_WECHAT_MP_APP_ID
+      || process.env.H5_WECHAT_APP_ID
+      || '',
+  });
+  const listenHost = process.env.MEMIND_CANARY_PROXY_HOST || '127.0.0.1';
+  const listenPort = Number(process.env.MEMIND_CANARY_PROXY_PORT ?? 18080);
+  const proxy = createMemindCanaryProxy({
+    stableTarget: process.env.MEMIND_CANARY_STABLE_URL || 'http://127.0.0.1:8081',
+    candidateTarget:
+      process.env.MEMIND_CANARY_CANDIDATE_URL || 'http://127.0.0.1:18081',
+    policy,
+    identityResolver,
+    diagnosticSecret: process.env.MEMIND_CANARY_DIAGNOSTIC_SECRET,
+  });
+
+  const shutdown = async () => {
+    proxy.server.close();
+    await pool.end().catch(() => {});
+  };
+  process.once('SIGTERM', shutdown);
+  process.once('SIGINT', shutdown);
+  proxy.server.listen(listenPort, listenHost, () => {
+    console.log(
+      `Memind canary proxy listening on http://${listenHost}:${listenPort} `
+      + `(stable=8081 candidate=18081 selectors=${JSON.stringify(selectorSummary)})`,
+    );
+  });
+}
+
+if (
+  process.argv[1]
+  && import.meta.url === pathToFileURL(process.argv[1]).href
+) {
+  main().catch((error) => {
+    console.error(error instanceof Error ? error.message : String(error));
+    process.exit(1);
+  });
+}