memind/server/portal-user-model-routes.mjs

import { getActiveSnapshot, getSnapshotInfo } from '../user-model-service/snapshot.mjs';
import { processIngestBatch } from '../user-model-service/service.mjs';
import { resolveCanonicalUserId } from '../user-model-service/canonical-user.mjs';

function assertRouter(api) {
  if (!api || typeof api.get !== 'function' || typeof api.post !== 'function') {
    throw new Error('attachPortalUserModelRoutes requires an Express-compatible router');
  }
}

function parseBearerToken(req) {
  const auth = req.headers.authorization ?? '';
  return auth.startsWith('Bearer ') ? auth.slice(7).trim() : '';
}

function createResolveUmsUser(getUserAuth) {
  return async function resolveUmsUser(req, _res, next) {
    if (req.currentUser?.id) return next();
    const bearer = parseBearerToken(req);
    const userAuth = getUserAuth?.();
    if (!bearer || !userAuth?.getMe) return next();
    try {
      const me = await userAuth.getMe(bearer);
      if (me) req.currentUser = me;
    } catch (err) {
      console.warn('[ums] bearer session verify failed:', err instanceof Error ? err.message : err);
    }
    return next();
  };
}

function createIngestAuth(getUserAuth) {
  return function ingestAuth(req, res, next) {
    const serviceToken = process.env.UMS_INGEST_TOKEN ?? '';
    const bearer = parseBearerToken(req);
    if (serviceToken && bearer === serviceToken) return next();
    if (req.currentUser?.id) return next();
    if (bearer && getUserAuth?.()?.verify) {
      return getUserAuth()
        .verify(bearer)
        .then((session) => {
          if (session?.userId) return next();
          return res.status(401).json({ error: 'unauthenticated' });
        })
        .catch(() => res.status(401).json({ error: 'unauthenticated' }));
    }
    return res.status(401).json({ error: 'unauthenticated' });
  };
}

function createRequireUser() {
  return function requireUser(req, res, next) {
    if (!req.currentUser?.id) {
      return res.status(401).json({ message: '未授权，请重新登录' });
    }
    return next();
  };
}

export function attachPortalUserModelRoutes(api, { getUmsPool, getUserAuth }) {
  assertRouter(api);
  const resolveUmsUser = createResolveUmsUser(getUserAuth);
  const ingestAuth = createIngestAuth(getUserAuth);
  const requireUser = createRequireUser();

  api.use('/v1/user-model', resolveUmsUser);

  api.post('/v1/user-model/ingest', ingestAuth, async (req, res) => {
    try {
      const pool = getUmsPool?.();
      if (!pool) return res.status(503).json({ error: 'user model service not configured' });
      const result = await processIngestBatch(pool, {
        items: req.body?.items ?? [],
        source_type: req.body?.source_type,
        dry_run: Boolean(req.body?.dry_run),
      });
      return res.json(result);
    } catch (err) {
      console.error('[ums] ingest failed', err);
      return res.status(500).json({ error: err instanceof Error ? err.message : 'ingest failed' });
    }
  });

  api.get('/v1/user-model/snapshot/info', requireUser, async (req, res) => {
    try {
      const pool = getUmsPool?.();
      if (!pool) return res.status(503).json({ error: 'user model service not configured' });
      const userId = resolveCanonicalUserId(req.currentUser.id);
      const projection = String(req.query.projection ?? 'default');
      const info = await getSnapshotInfo(pool, userId, projection);
      if (!info) return res.status(404).json({ error: 'snapshot_not_ready' });
      if (req.headers['if-none-match'] === info.content_hash) return res.status(304).end();
      return res.json(info);
    } catch (err) {
      console.error('[ums] snapshot info failed', err);
      return res.status(500).json({ error: err instanceof Error ? err.message : 'snapshot info failed' });
    }
  });

  api.get('/v1/user-model/snapshot', requireUser, async (req, res) => {
    try {
      const pool = getUmsPool?.();
      if (!pool) return res.status(503).json({ error: 'user model service not configured' });
      const userId = resolveCanonicalUserId(req.currentUser.id);
      const projection = String(req.query.projection ?? 'default');
      const snap = await getActiveSnapshot(pool, userId, projection);
      if (!snap) return res.status(404).json({ error: 'snapshot_not_ready' });
      if (req.headers['if-none-match'] === snap.content_hash) return res.status(304).end();
      res.setHeader('ETag', snap.content_hash);
      return res.json({
        snapshot_id: snap.snapshot_id,
        user_id: snap.user_id,
        projection: snap.projection,
        profile_version: snap.profile_version,
        fast_revision: snap.fast_revision,
        content_hash: snap.content_hash,
        byte_size: snap.byte_size,
        stale_after_sec: snap.stale_after_sec,
        core: snap.core,
        meta: snap.meta,
      });
    } catch (err) {
      console.error('[ums] snapshot failed', err);
      return res.status(500).json({ error: err instanceof Error ? err.message : 'snapshot failed' });
    }
  });

  api.get('/v1/user-model/candidates', requireUser, async (req, res) => {
    try {
      const pool = getUmsPool?.();
      if (!pool) return res.status(503).json({ error: 'user model service not configured' });
      const userId = resolveCanonicalUserId(req.currentUser.id);
      const status = String(req.query.status ?? 'open,accepted');
      const statuses = status.split(',').map((s) => s.trim()).filter(Boolean);
      const placeholders = statuses.map(() => '?').join(',');
      const [rows] = await pool.query(
        `SELECT candidate_id, candidate_type, hypothesis_json, status, promotion_score, confidence,
                signal_ids, evidence_ids, first_seen_at, last_seen_at
         FROM um_candidates
         WHERE user_id = ? AND status IN (${placeholders})
         ORDER BY promotion_score DESC
         LIMIT ?`,
        [userId, ...statuses, Math.min(100, Number(req.query.limit ?? 50))],
      );
      return res.json({
        items: rows.map((row) => ({
          candidate_id: row.candidate_id,
          candidate_type: row.candidate_type,
          hypothesis:
            typeof row.hypothesis_json === 'string'
              ? JSON.parse(row.hypothesis_json)
              : row.hypothesis_json,
          status: row.status,
          promotion_score: Number(row.promotion_score),
          confidence: Number(row.confidence),
          signal_ids:
            typeof row.signal_ids === 'string' ? JSON.parse(row.signal_ids) : row.signal_ids,
          evidence_ids:
            typeof row.evidence_ids === 'string' ? JSON.parse(row.evidence_ids) : row.evidence_ids,
        })),
      });
    } catch (err) {
      console.error('[ums] candidates failed', err);
      return res.status(500).json({ error: err instanceof Error ? err.message : 'candidates failed' });
    }
  });
}