memind/server/portal-session-routes.mjs

import fs from 'node:fs';
import path from 'node:path';
import {
  isDirectChatSessionId,
  isPortalDirectChatSnapshot,
  sendDirectChatSessionEvents,
  shouldExpirePortalDirectChatSnapshot,
} from '../direct-chat-service.mjs';
import {
  resolveAnalyticsOwnerLabel,
  resolveAnalyticsOwnerSegment,
  sendMindSpaceAnalyticsEvent,
} from '../mindspace-analytics.mjs';
import {
  sendMindSpaceRybbitEvent,
} from '../mindspace-rybbit.mjs';
import {
  markPageDeliveryContractReady,
  preparePageDeliveryContract,
} from '../mindspace-delivery-contract.mjs';
import { maybeRepairH5HtmlAfterFinish } from '../mindspace-h5-html-finish-guard.mjs';
import { maybeRepairPageDataAfterFinish } from '../mindspace-page-data-finish-guard.mjs';
import {
  collectPublicHtmlWritePathsFromSessionEvent,
  materializePublicHtmlWritesFromSessionEvent,
  syncPublicHtmlAfterFinish,
} from '../mindspace-public-finish-sync.mjs';
import {
  resolveMindSpaceRuntimeConfig,
  resolveMindSpaceUserPublishDir,
} from '../mindspace-runtime-config.mjs';
import {
  repairSessionConversationFromDb,
} from '../conversation-repair.mjs';
import { filterNonemptyUserVisibleMessages } from '../conversation-transcript-persist.mjs';
import { sanitizeSessionConversationPublicHtmlLinks } from '../tkmind-proxy.mjs';
import { PUBLISH_ROOT_DIR } from '../user-publish.mjs';

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

export function attachPortalSessionRoutes(
  api,
  {
    h5Root = '',
    env = process.env,
    waitForUserAuthReady = async () => {},
    getUserAuth = () => null,
    getTkmindProxy = () => null,
    getSessionSnapshotService = () => null,
    getAuthPool = () => null,
    getMindSpaceAssets = () => null,
    getMemoryV2 = () => null,
    getMindSpaceAnalyticsConfig = () => null,
    getMindSpaceRybbitConfig = () => null,
    isWorkspaceMaintenanceEnabled = () => false,
    ownsAgentSession,
    unregisterAgentSessionForUser,
    beginSessionPageDelivery,
    endSessionPageDelivery,
    syncUserGeneratedPages,
    registerPublicHtmlArtifactsForConversation,
    isDirectChatSessionIdFn = isDirectChatSessionId,
    isPortalDirectChatSnapshotFn = isPortalDirectChatSnapshot,
    sendDirectChatSessionEventsFn = sendDirectChatSessionEvents,
    shouldExpirePortalDirectChatSnapshotFn =
      shouldExpirePortalDirectChatSnapshot,
    sanitizeSessionConversationFn =
      sanitizeSessionConversationPublicHtmlLinks,
    repairSessionConversationFromDbFn =
      repairSessionConversationFromDb,
    filterNonemptyUserVisibleMessagesFn =
      filterNonemptyUserVisibleMessages,
    resolveMindSpaceUserPublishDirFn =
      resolveMindSpaceUserPublishDir,
    collectPublicHtmlWritePathsFn =
      collectPublicHtmlWritePathsFromSessionEvent,
    preparePageDeliveryContractFn =
      preparePageDeliveryContract,
    materializePublicHtmlWritesFn =
      materializePublicHtmlWritesFromSessionEvent,
    sendMindSpaceAnalyticsEventFn =
      sendMindSpaceAnalyticsEvent,
    sendMindSpaceRybbitEventFn =
      sendMindSpaceRybbitEvent,
    resolveAnalyticsOwnerSegmentFn =
      resolveAnalyticsOwnerSegment,
    resolveAnalyticsOwnerLabelFn =
      resolveAnalyticsOwnerLabel,
    syncPublicHtmlAfterFinishFn = syncPublicHtmlAfterFinish,
    resolveMindSpaceRuntimeConfigFn =
      resolveMindSpaceRuntimeConfig,
    maybeRepairH5HtmlAfterFinishFn =
      maybeRepairH5HtmlAfterFinish,
    maybeRepairPageDataAfterFinishFn =
      maybeRepairPageDataAfterFinish,
    markPageDeliveryContractReadyFn =
      markPageDeliveryContractReady,
    fileExists = fs.existsSync,
    resolvePath = path.resolve,
    publishRootDir = PUBLISH_ROOT_DIR,
    logger = console,
  } = {},
) {
  assertRouter(api);
  if (
    typeof ownsAgentSession !== 'function' ||
    typeof unregisterAgentSessionForUser !== 'function' ||
    typeof beginSessionPageDelivery !== 'function' ||
    typeof endSessionPageDelivery !== 'function' ||
    typeof syncUserGeneratedPages !== 'function' ||
    typeof registerPublicHtmlArtifactsForConversation !== 'function'
  ) {
    throw new Error(
      'attachPortalSessionRoutes requires session route dependencies',
    );
  }

  // Session detail — serve from DB snapshot cache when fresh, fall through to Goose on miss.
  api.get('/sessions/:sessionId', async (req, res, next) => {
    await waitForUserAuthReady();
    const userAuth = getUserAuth();
    const tkmindProxy = getTkmindProxy();
    if (!userAuth || !tkmindProxy) return next();
    const sessionId = req.params.sessionId;
    const owns = await ownsAgentSession(
      req.currentUser.id,
      sessionId,
    );
    if (!owns) {
      return res.status(403).json({ message: '无权访问该会话' });
    }
    // Hints from the client (session list already has these values).
    const hintMc = req.query.hint_mc
      ? Number(req.query.hint_mc)
      : null;
    const hintUa = req.query.hint_ua
      ? String(req.query.hint_ua)
      : null;
    const snapshotService = getSessionSnapshotService();
    const authPool = getAuthPool();

    try {
      if (snapshotService?.isEnabled()) {
        let snapshot = await snapshotService.get(sessionId);
        if (snapshot) {
          if (
            authPool &&
            (await shouldExpirePortalDirectChatSnapshotFn(
              authPool,
              sessionId,
              snapshot,
            ))
          ) {
            await snapshotService.remove(sessionId).catch(() => {});
            snapshot = null;
          }
        }
        if (snapshot) {
          // REGRESSION GUARD: without both hints, stale snapshot can wipe mid-turn chat.
          const canUseSnapshotCache =
            hintMc != null && hintUa != null;
          const mcMatch =
            snapshot.meta.synced_msg_count === hintMc;
          const uaMatch =
            snapshot.meta.source_updated_at === hintUa;
          if (
            isDirectChatSessionIdFn(sessionId) ||
            isPortalDirectChatSnapshotFn(snapshot, { sessionId }) ||
            (canUseSnapshotCache && mcMatch && uaMatch)
          ) {
            const sanitizedMessages =
              sanitizeSessionConversationFn(
                snapshot.messages,
                req.currentUser,
              );
            // Cache hit — reconstruct a Goose-compatible session response.
            let cachedGooseSession = {
              ...snapshot.session,
              // Embed only userVisible messages so getSession callers still work.
              conversation: sanitizedMessages,
            };
            if (authPool) {
              cachedGooseSession =
                await repairSessionConversationFromDbFn(
                  authPool,
                  cachedGooseSession,
                  sessionId,
                  req.currentUser.id,
                );
            }
            return res.json(cachedGooseSession);
          }
        }
      }
    } catch {
      // Snapshot read error: fall through silently to Goose.
    }

    // Cache miss — proxy to Goose and write-through on success.
    try {
      const target = await tkmindProxy.resolveTarget(sessionId);
      const upstream = await tkmindProxy.apiFetchTo(
        target,
        `/sessions/${encodeURIComponent(sessionId)}`,
        { method: 'GET' },
      );
      if (!upstream.ok) {
        const text = await upstream.text().catch(() => '');
        return res.status(upstream.status).send(text);
      }
      let gooseSession = await upstream.json();
      if (Array.isArray(gooseSession.conversation)) {
        gooseSession.conversation =
          sanitizeSessionConversationFn(
            gooseSession.conversation,
            req.currentUser,
          );
      }
      if (authPool) {
        gooseSession = await repairSessionConversationFromDbFn(
          authPool,
          gooseSession,
          sessionId,
          req.currentUser.id,
        );
      }
      // Write-through: persist snapshot async, don't block the response.
      if (snapshotService?.isEnabled()) {
        const messages = filterNonemptyUserVisibleMessagesFn(
          gooseSession.conversation ?? [],
        );
        if (messages.length > 0) {
          void snapshotService
            .save(
              sessionId,
              req.currentUser.id,
              gooseSession,
              messages,
            )
            .catch(() => {});
        }
      }
      return res.json(gooseSession);
    } catch (error) {
      return res.status(502).json({
        message:
          error instanceof Error
            ? error.message
            : '读取会话失败',
      });
    }
  });

  api.delete('/sessions/:sessionId', async (req, res, next) => {
    await waitForUserAuthReady();
    const userAuth = getUserAuth();
    const tkmindProxy = getTkmindProxy();
    if (!userAuth || !tkmindProxy) return next();
    const sessionId = req.params.sessionId;
    const owns = await ownsAgentSession(
      req.currentUser.id,
      sessionId,
    );
    if (!owns) {
      return res.status(403).json({ message: '无权访问该会话' });
    }
    const snapshotService = getSessionSnapshotService();
    try {
      if (isDirectChatSessionIdFn(sessionId)) {
        await unregisterAgentSessionForUser(
          req.currentUser.id,
          sessionId,
        );
        void snapshotService?.remove(sessionId).catch(() => {});
        return res.status(204).end();
      }
      const deleteTarget =
        await tkmindProxy.resolveTarget(sessionId);
      const upstream = await tkmindProxy.apiFetchTo(
        deleteTarget,
        `/sessions/${encodeURIComponent(sessionId)}`,
        { method: 'DELETE' },
      );
      if (!upstream.ok && upstream.status !== 404) {
        const text = await upstream.text().catch(() => '');
        return res
          .status(upstream.status)
          .send(text || '删除会话失败');
      }
      await unregisterAgentSessionForUser(
        req.currentUser.id,
        sessionId,
      );
      // Remove snapshot so it doesn't linger after deletion.
      void snapshotService?.remove(sessionId).catch(() => {});
      return res.status(204).end();
    } catch (error) {
      return res.status(500).json({
        message:
          error instanceof Error
            ? error.message
            : '删除会话失败',
      });
    }
  });

  api.get(
    '/sessions/:sessionId/events',
    async (req, res, next) => {
      await waitForUserAuthReady();
      const userAuth = getUserAuth();
      const tkmindProxy = getTkmindProxy();
      if (!userAuth || !tkmindProxy) return next();
      const sessionId = req.params.sessionId;
      const owns = await ownsAgentSession(
        req.currentUser.id,
        sessionId,
      );
      if (!owns) {
        return res
          .status(403)
          .json({ message: '无权访问该会话' });
      }
      const snapshotService = getSessionSnapshotService();
      if (isDirectChatSessionIdFn(sessionId)) {
        const snapshot = await snapshotService
          ?.get(sessionId)
          .catch(() => null);
        if (!snapshot) {
          return res.status(404).json({ message: '会话不存在' });
        }
        return sendDirectChatSessionEventsFn(req, res, snapshot);
      }
      const portalDirectSnapshot = await snapshotService
        ?.get(sessionId)
        .catch(() => null);
      const authPool = getAuthPool();
      if (
        portalDirectSnapshot &&
        authPool &&
        (await shouldExpirePortalDirectChatSnapshotFn(
          authPool,
          sessionId,
          portalDirectSnapshot,
        ))
      ) {
        await snapshotService
          ?.remove(sessionId)
          .catch(() => {});
      } else if (
        portalDirectSnapshot &&
        isPortalDirectChatSnapshotFn(
          portalDirectSnapshot,
          { sessionId },
        )
      ) {
        return sendDirectChatSessionEventsFn(
          req,
          res,
          portalDirectSnapshot,
        );
      }
      const publishDir = resolveMindSpaceUserPublishDirFn(
        h5Root,
        { id: req.currentUser.id },
      );
      // `proxySessionEvents` deliberately invokes `onEvent` synchronously so an
      // async callback here would leave rejected database writes unhandled.
      // Retain every in-flight contract write and await it before marking files
      // deliverable after Finish.
      const deliveryContractWrites = new Map();
      const generationAnalyticsEvents = new Set();
      const syncPublicHtmlDuringStream = (event) => {
        const paths = collectPublicHtmlWritePathsFn(event, {
          publishDir,
        });
        const eventMessages =
          event?.type === 'Message' && event.message
            ? [event.message]
            : event?.type === 'UpdateConversation' &&
                Array.isArray(event.conversation)
              ? event.conversation
              : [];
        const pgRequired = eventMessages.some(
          (message) =>
            message?.role === 'user' &&
            message?.metadata?.memindRun?.pgRequired === true,
        );
        for (const relativePath of paths) {
          if (deliveryContractWrites.has(relativePath)) continue;
          const write = preparePageDeliveryContractFn({
            pool: authPool,
            userId: req.currentUser.id,
            requestId: sessionId,
            relativePath,
            pgRequired,
          }).catch((error) => {
            logger.warn(
              `[MindSpace] failed to prepare delivery contract for ${relativePath}: ${error?.message || error}`,
            );
            return null;
          });
          deliveryContractWrites.set(relativePath, write);
        }
        materializePublicHtmlWritesFn(event, { publishDir });
        for (const relativePath of paths) {
          const absolutePath = resolvePath(
            publishDir,
            relativePath,
          );
          if (
            generationAnalyticsEvents.has(relativePath) ||
            !fileExists(absolutePath)
          ) {
            continue;
          }
          generationAnalyticsEvents.add(relativePath);
          const analyticsPayload = {
            eventName: 'page_generated',
            ownerId: req.currentUser.id,
            ownerSegment:
              resolveAnalyticsOwnerSegmentFn(req.currentUser),
            ownerLabel:
              resolveAnalyticsOwnerLabelFn(req.currentUser),
            pageId: relativePath,
            publicationId: sessionId,
            agentRunId: sessionId,
            channel: 'h5',
            url:
              `/${publishRootDir}/${req.currentUser.id}/${relativePath}`,
          };
          void sendMindSpaceAnalyticsEventFn({
            config: getMindSpaceAnalyticsConfig(),
            ...analyticsPayload,
          });
          void sendMindSpaceRybbitEventFn({
            config: getMindSpaceRybbitConfig(),
            ...analyticsPayload,
          });
        }
      };
      // After Finish, refresh the snapshot and persist any newly generated public
      // workspace HTML into the asset store before a later restart rebuilds the
      // workspace from DB-backed assets only.
      const onAfterFinish = async (sid, uid) => {
        beginSessionPageDelivery(sid);
        try {
          const apiFetchFn = async (pathname, init) => {
            const target = await tkmindProxy.resolveTarget(sid);
            return tkmindProxy.apiFetchTo(
              target,
              pathname,
              init,
            );
          };
          let messages = null;
          if (snapshotService?.isEnabled()) {
            await snapshotService.refresh(
              sid,
              uid,
              apiFetchFn,
            );
            messages =
              (await snapshotService.get(sid))?.messages ?? null;
          } else {
            try {
              const upstream = await apiFetchFn(
                `/sessions/${encodeURIComponent(sid)}`,
                { method: 'GET' },
              );
              if (upstream.ok) {
                const payload = await upstream
                  .json()
                  .catch(() => null);
                messages = Array.isArray(payload?.conversation)
                  ? payload.conversation.filter(
                      (message) =>
                        message?.metadata?.userVisible,
                    )
                  : null;
              }
            } catch {
              messages = null;
            }
          }
          const { storageRoot } =
            resolveMindSpaceRuntimeConfigFn(h5Root, env);
          const syncResult = await syncPublicHtmlAfterFinishFn({
            messages,
            currentUser: req.currentUser,
            publishDir,
            sessionId: sid,
            pool: authPool,
            storageRoot,
            h5Root,
            syncWorkspaceAssets:
              isWorkspaceMaintenanceEnabled() &&
              getMindSpaceAssets()
                ? (userId, options) =>
                    getMindSpaceAssets().syncWorkspaceAssets(
                      userId,
                      options,
                    )
                : null,
            registerPublicHtmlArtifacts: (_userId, options) =>
              registerPublicHtmlArtifactsForConversation({
                user: req.currentUser,
                publishDir,
                sessionId: options?.sessionId,
                relativePaths: options?.relativePaths,
                artifactRefs: options?.artifactRefs,
              }),
          });
          if (
            Array.isArray(syncResult?.docxSync?.missing) &&
            syncResult.docxSync.missing.length > 0
          ) {
            logger.warn(
              `[MindSpace] missing public download files after finish for user ${uid}: ${syncResult.docxSync.missing.join(', ')}`,
            );
          }
          const htmlDelivery =
            await maybeRepairH5HtmlAfterFinishFn({
              sessionId: sid,
              userId: uid,
              currentUser: req.currentUser,
              messages,
              publishDir,
              syncResult,
              tkmindProxy,
            });
          const lastUserMessage = [
            ...(Array.isArray(messages) ? messages : []),
          ]
            .reverse()
            .find((message) => message?.role === 'user');
          const lastUserText =
            typeof lastUserMessage?.content === 'string'
              ? lastUserMessage.content
              : Array.isArray(lastUserMessage?.content)
                ? lastUserMessage.content
                    .filter((item) => item?.type === 'text')
                    .map((item) =>
                      String(item.text ?? '').trim(),
                    )
                    .filter(Boolean)
                    .join('\n')
                : '';
          await syncUserGeneratedPages(uid, {
            sessionId: sid,
          });
          const pageDataDelivery =
            await maybeRepairPageDataAfterFinishFn({
              sessionId: sid,
              userId: uid,
              publishDir,
              messages,
              pool: authPool,
              h5Root,
              storageRoot,
              tkmindProxy,
              userText: lastUserText,
            });
          const htmlReady = htmlDelivery?.skipped === 'ok';
          const pageDataReady = [
            'ok',
            'not_page_data',
          ].includes(
            String(pageDataDelivery?.skipped ?? ''),
          );
          if (htmlReady && pageDataReady) {
            const publicHtmlRelativePaths =
              [
                ...new Set([
                  ...(syncResult?.publicHtmlRelativePaths ??
                    []),
                  ...deliveryContractWrites.keys(),
                ]),
              ].sort();
            const pgRequired = [
              ...(Array.isArray(messages) ? messages : []),
            ].some(
              (message) =>
                message?.role === 'user' &&
                message?.metadata?.memindRun?.pgRequired ===
                  true,
            );
            for (const relativePath of publicHtmlRelativePaths) {
              // A Finish-only write may not have reached the stream callback. This
              // also upgrades an early partial stream contract with the definitive
              // user delivery choice before it becomes ready.
              await (
                deliveryContractWrites.get(relativePath) ??
                preparePageDeliveryContractFn({
                  pool: authPool,
                  userId: uid,
                  requestId: sid,
                  relativePath,
                  pgRequired,
                })
              );
              if (
                deliveryContractWrites.has(relativePath)
              ) {
                await preparePageDeliveryContractFn({
                  pool: authPool,
                  userId: uid,
                  requestId: sid,
                  relativePath,
                  pgRequired,
                });
              }
              await markPageDeliveryContractReadyFn({
                pool: authPool,
                userId: uid,
                relativePath,
              }).catch(() => false);
            }
          }
          const memoryV2 = getMemoryV2();
          if (
            lastUserMessage &&
            memoryV2?.observePersonalMemory
          ) {
            await memoryV2
              .observePersonalMemory({
                userId: uid,
                sessionId: sid,
                messages: [lastUserMessage],
              })
              .catch((error) => {
                logger.warn(
                  `[memory-v2] finish shadow observation skipped for session ${sid}: ${error instanceof Error ? error.message : error}`,
                );
              });
          }
        } finally {
          endSessionPageDelivery(sid);
        }
      };
      return tkmindProxy.proxySessionEvents(
        req,
        res,
        sessionId,
        {
          onAfterFinish,
          onEvent: syncPublicHtmlDuringStream,
        },
      );
    },
  );
}