import {
  isDirectChatSessionId,
  isPortalDirectChatSnapshot,
  sendDirectChatSessionEvents,
  shouldExpirePortalDirectChatSnapshot,
} from '../direct-chat-service.mjs';
import {
  resolveAnalyticsOwnerLabel,
  resolveAnalyticsOwnerSegment,
  resolveAnalyticsPlan,
  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 { hasPublicHtmlWriteRequestInSessionEvent } from '../mindspace-public-finish-sync.mjs';
import {
  repairSessionConversationFromDb,
} from '../conversation-repair.mjs';
import { filterNonemptyUserVisibleMessages } from '../conversation-transcript-persist.mjs';
import { sanitizeSessionConversationPublicHtmlLinks } from '../tkmind-proxy.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,
  {
    waitForUserAuthReady = async () => {},
    getUserAuth = () => null,
    getTkmindProxy = () => null,
    getSessionSnapshotService = () => null,
    getAuthPool = () => null,
    getMindSpacePublicFinish = () => null,
    getMemoryV2 = () => null,
    getMindSpaceAnalyticsConfig = () => null,
    getMindSpaceRybbitConfig = () => null,
    ownsAgentSession,
    unregisterAgentSessionForUser,
    beginSessionPageDelivery,
    endSessionPageDelivery,
    syncUserGeneratedPages,
    isDirectChatSessionIdFn = isDirectChatSessionId,
    isPortalDirectChatSnapshotFn = isPortalDirectChatSnapshot,
    sendDirectChatSessionEventsFn = sendDirectChatSessionEvents,
    shouldExpirePortalDirectChatSnapshotFn =
      shouldExpirePortalDirectChatSnapshot,
    sanitizeSessionConversationFn =
      sanitizeSessionConversationPublicHtmlLinks,
    repairSessionConversationFromDbFn =
      repairSessionConversationFromDb,
    filterNonemptyUserVisibleMessagesFn =
      filterNonemptyUserVisibleMessages,
    preparePageDeliveryContractFn =
      preparePageDeliveryContract,
    hasPublicHtmlWriteRequestInSessionEventFn =
      hasPublicHtmlWriteRequestInSessionEvent,
    sendMindSpaceAnalyticsEventFn =
      sendMindSpaceAnalyticsEvent,
    sendMindSpaceRybbitEventFn =
      sendMindSpaceRybbitEvent,
    resolveAnalyticsOwnerSegmentFn =
      resolveAnalyticsOwnerSegment,
    resolveAnalyticsOwnerLabelFn =
      resolveAnalyticsOwnerLabel,
    resolveAnalyticsPlanFn = resolveAnalyticsPlan,
    maybeRepairH5HtmlAfterFinishFn =
      maybeRepairH5HtmlAfterFinish,
    maybeRepairPageDataAfterFinishFn =
      maybeRepairPageDataAfterFinish,
    markPageDeliveryContractReadyFn =
      markPageDeliveryContractReady,
    finishDeliveryRetryDelaysMs = [250, 1_000],
    finishDeliveryRetryWaitFn = (delayMs) =>
      new Promise((resolve) =>
        setTimeout(resolve, delayMs),
      ),
    logger = console,
  } = {},
) {
  assertRouter(api);
  if (
    typeof ownsAgentSession !== 'function' ||
    typeof unregisterAgentSessionForUser !== 'function' ||
    typeof beginSessionPageDelivery !== 'function' ||
    typeof endSessionPageDelivery !== 'function' ||
    typeof syncUserGeneratedPages !== '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,
        );
      }
      // `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 streamPublicHtmlWrites = new Set();
      const generationAnalyticsEvents = new Set();
      const syncPublicHtmlDuringStream = (event) => {
        if (
          !hasPublicHtmlWriteRequestInSessionEventFn(event)
        ) {
          return;
        }
        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,
        );
        const write = Promise.resolve()
          .then(async () => {
            const publicFinish =
              getMindSpacePublicFinish();
            if (!publicFinish) {
              throw new Error(
                'MindSpace public finish service unavailable',
              );
            }
            const result =
              await publicFinish.materializeSessionEvent({
                userId: req.currentUser.id,
                event,
              });
            const artifactsByPath = new Map(
              (
                Array.isArray(result?.publicHtmlArtifacts)
                  ? result.publicHtmlArtifacts
                  : []
              ).map((artifact) => [
                artifact.relativePath,
                artifact,
              ]),
            );
            const paths = Array.isArray(
              result?.publicHtmlRelativePaths,
            )
              ? result.publicHtmlRelativePaths
              : [];
            for (const relativePath of paths) {
              if (
                !deliveryContractWrites.has(relativePath)
              ) {
                const contractWrite =
                  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,
                  contractWrite,
                );
              }
              const canonicalUrl =
                artifactsByPath.get(relativePath)
                  ?.canonicalUrl;
              if (
                generationAnalyticsEvents.has(
                  relativePath,
                ) ||
                !canonicalUrl
              ) {
                continue;
              }
              generationAnalyticsEvents.add(relativePath);
              const analyticsPayload = {
                eventName: 'page_generated',
                ownerId: req.currentUser.id,
                ownerSegment:
                  resolveAnalyticsOwnerSegmentFn(
                    req.currentUser,
                  ),
                ownerLabel:
                  resolveAnalyticsOwnerLabelFn(
                    req.currentUser,
                  ),
                planType:
                  resolveAnalyticsPlanFn(
                    req.currentUser,
                  ),
                generatedAt: new Date().toISOString(),
                pageId: relativePath,
                publicationId: sessionId,
                agentRunId: sessionId,
                channel: 'h5',
                url: canonicalUrl,
              };
              void sendMindSpaceAnalyticsEventFn({
                config:
                  getMindSpaceAnalyticsConfig(),
                ...analyticsPayload,
              });
              void sendMindSpaceRybbitEventFn({
                config:
                  getMindSpaceRybbitConfig(),
                ...analyticsPayload,
              });
            }
          })
          .catch((error) => {
            logger.warn(
              `[MindSpace] failed to materialize streamed public HTML for ${sessionId}: ${error?.message || error}`,
            );
          });
        streamPublicHtmlWrites.add(write);
        void write.finally(() => {
          streamPublicHtmlWrites.delete(write);
        });
      };
      // 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 finalizeAfterFinishOnce = 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;
            }
          }
          await Promise.all([...streamPublicHtmlWrites]);
          const publicFinish =
            getMindSpacePublicFinish();
          if (!publicFinish) {
            throw new Error(
              'MindSpace public finish service unavailable',
            );
          }
          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')
                : '';
          const syncResult =
            await publicFinish.syncAfterFinish({
              userId: uid,
              sessionId: sid,
              messages,
              currentUser: {
                id: req.currentUser.id,
                username:
                  req.currentUser.username,
              },
            });
          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,
              syncResult,
              evaluation:
                syncResult
                  ?.deliveryEvaluation,
              tkmindProxy,
            });
          await syncUserGeneratedPages(uid, {
            sessionId: sid,
          });
          const pageDataPreparation =
            await publicFinish.preparePageDataAfterFinish({
              userId: uid,
              messages,
              userText: lastUserText,
            });
          const pageDataDelivery =
            await maybeRepairPageDataAfterFinishFn({
              sessionId: sid,
              userId: uid,
              messages,
              tkmindProxy,
              userText: lastUserText,
              preparation: pageDataPreparation,
            });
          const htmlReady = htmlDelivery?.skipped === 'ok';
          const pageDataReady = [
            'ok',
            'not_page_data',
          ].includes(
            String(pageDataDelivery?.skipped ?? ''),
          );
          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,
          );
          if (
            publicHtmlRelativePaths.length > 0 &&
            (!htmlReady || !pageDataReady)
          ) {
            throw new Error(
              'page delivery guards are not ready: '
                + `html=${htmlDelivery?.skipped ?? 'unknown'} `
                + `pageData=${pageDataDelivery?.skipped ?? 'unknown'}`,
            );
          }
          if (htmlReady && pageDataReady) {
            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);
        }
      };
      const onAfterFinish = async (sid, uid) => {
        const retryDelays = Array.isArray(
          finishDeliveryRetryDelaysMs,
        )
          ? finishDeliveryRetryDelaysMs
              .map((delayMs) =>
                Math.max(0, Number(delayMs) || 0),
              )
          : [];
        let lastError = null;
        for (
          let attempt = 0;
          attempt <= retryDelays.length;
          attempt += 1
        ) {
          if (attempt > 0) {
            await finishDeliveryRetryWaitFn(
              retryDelays[attempt - 1],
            );
          }
          try {
            return await finalizeAfterFinishOnce(
              sid,
              uid,
            );
          } catch (error) {
            lastError = error;
            logger.warn(
              `[MindSpace] Finish delivery finalization failed for session ${sid} `
                + `(attempt ${attempt + 1}/${retryDelays.length + 1}): `
                + `${error instanceof Error ? error.message : error}`,
            );
          }
        }
        throw lastError;
      };
      return tkmindProxy.proxySessionEvents(
        req,
        res,
        sessionId,
        {
          onAfterFinish,
          onEvent: syncPublicHtmlDuringStream,
        },
      );
    },
  );
}