memind/server/portal-integration-services-bootstrap.mjs

import {
  resolveAnalyticsOwnerLabel,
  resolveAnalyticsOwnerSegment,
  resolveAnalyticsPlan,
  sendMindSpaceAnalyticsEvent,
} from '../mindspace-analytics.mjs';
import { createPageEditSessionService } from '../mindspace-page-edit-session.mjs';
import { createNotificationDispatcher } from '../notification-dispatcher.mjs';
import { startScheduleReminderWorker } from '../schedule-reminder-worker.mjs';
import { startScheduledTaskWorker } from '../scheduled-task-worker.mjs';
import { isScheduledTaskWorkerEnabled } from '../scheduled-task-worker-config.mjs';
import { startHealthBaselineWorker } from '../health-baseline-worker.mjs';
import { startWechatNewsMorningDraftWorker } from '../wechat-news-morning-draft-worker.mjs';
import { createWechatNewsMorningDraftService } from '../wechat-news-morning-draft.mjs';
import { createHealthEventNotificationService } from '../health-event-notification-service.mjs';
import { isPassiveCanaryRuntime } from './portal-runtime-role.mjs';
import { loadWechatMpModule } from '../wechat-mp-loader.mjs';
import { createToolGateway } from '../tool-gateway.mjs';
import {
  createPageDataDeliveryCodeReviewService,
} from '../page-data-delivery-code-review.mjs';

export async function bootstrapPortalIntegrationServices({
  pool,
  h5Root,
  codeRoot = h5Root,
  env = process.env,
  usersRoot,
  wechatMpConfig,
  mindSpacePublicFinish,
  userAuth,
  sessionAccess,
  tkmindProxy,
  scheduleService,
  scheduledTaskService = null,
  intentDraftService = null,
  taskUnifiedService = null,
  sessionSnapshotService = null,
  wechatScheduleLlmConfigService,
  wechatSubscribeMorningLlmConfigService = null,
  wechatScheduledTaskManageLlmConfigService = null,
  wechatIntentRouter = null,
  wechatCursorExecutorPolicyService = null,
  agentRunGateway = null,
  llmProviderService,
  chatIntentRouter,
  systemDisclosurePolicyService,
  mindSpaceAnalyticsConfig,
  subscriptionService,
  apiTarget,
  apiSecret,
  mindSpacePages,
  mindSpacePageLiveEdit,
  logger = console,
  healthChannelStore = null,
  healthObservationStore = null,
  healthObservationService = null,
  healthDocumentStore = null,
  healthDataRuntime = null,
  healthEventStore = null,
  loadWechatMpModuleFn = loadWechatMpModule,
  resolveAnalyticsOwnerSegmentFn =
    resolveAnalyticsOwnerSegment,
  resolveAnalyticsOwnerLabelFn =
    resolveAnalyticsOwnerLabel,
  resolveAnalyticsPlanFn = resolveAnalyticsPlan,
  sendMindSpaceAnalyticsEventFn =
    sendMindSpaceAnalyticsEvent,
  createNotificationDispatcherFn =
    createNotificationDispatcher,
  startScheduleReminderWorkerFn =
    startScheduleReminderWorker,
  startScheduledTaskWorkerFn =
    startScheduledTaskWorker,
  startHealthBaselineWorkerFn =
    startHealthBaselineWorker,
  startWechatNewsMorningDraftWorkerFn =
    startWechatNewsMorningDraftWorker,
  createWechatNewsMorningDraftServiceFn =
    createWechatNewsMorningDraftService,
  createPageEditSessionServiceFn =
    createPageEditSessionService,
  createToolGatewayFn = createToolGateway,
  createPageDataDeliveryCodeReviewServiceFn =
    createPageDataDeliveryCodeReviewService,
  setIntervalFn = setInterval,
} = {}) {
  if (
    !pool ||
    !h5Root ||
    !usersRoot ||
    !userAuth ||
    !sessionAccess ||
    !tkmindProxy ||
    !llmProviderService
  ) {
    throw new Error(
      'bootstrapPortalIntegrationServices requires Portal integration dependencies',
    );
  }

  const wechatMpRoot = codeRoot || h5Root;
  logger.log?.(`[WeChat MP] Loading runtime module from code root: ${wechatMpRoot}`);
  const wechatMp = await loadWechatMpModuleFn(wechatMpRoot);
  const pageDataDeliveryReviewer =
    wechatMpConfig?.pageDataAiderReviewEnabled === true
      ? createPageDataDeliveryCodeReviewServiceFn({
          toolGateway: createToolGatewayFn({
            llmProviderService,
            env,
          }),
          userAuth,
          timeoutMs: Number(
            env.H5_WECHAT_MP_PAGE_DATA_AIDER_REVIEW_TIMEOUT_MS ??
              15 * 60 * 1000,
          ),
          logger,
        })
      : null;
  const wechatMpService =
    wechatMp.createWechatMpService({
      config: wechatMpConfig,
      userAuth,
      sessionAccess,
      h5Root,
      htmlDeliveryAuthority:
        typeof mindSpacePublicFinish
          ?.prepareWechatHtmlDelivery ===
          'function' &&
        typeof mindSpacePublicFinish
          ?.ensureWechatFreshPageThumbnails ===
          'function'
          ? mindSpacePublicFinish
          : null,
      pageDataFinishGuard:
        typeof mindSpacePublicFinish
          ?.prepareWechatPageDataDelivery ===
        'function'
          ? mindSpacePublicFinish
          : null,
      pageDataDeliveryReviewer,
      wechatCursorExecutorPolicyService,
      agentRunGateway,
      apiFetch: tkmindProxy.apiFetch,
      startAgentSession: ({
        userId,
        workingDir,
        sessionPolicy,
      }) =>
        tkmindProxy.startSessionForUser(userId, {
          workingDir,
          sessionPolicy,
          origin: 'wechat',
        }),
      sessionApiFetch: async (
        sessionId,
        pathname,
        init,
      ) => {
        const target =
          await tkmindProxy.resolveTarget(sessionId);
        return tkmindProxy.apiFetchTo(
          target,
          pathname,
          init,
        );
      },
      submitSessionReply: ({
        userId,
        sessionId,
        requestId,
        userMessage,
        options,
      }) =>
        tkmindProxy.submitSessionReplyForUser(
          userId,
          sessionId,
          requestId,
          userMessage,
          options,
        ),
      scheduleService:
        env.H5_SCHEDULE_ENABLED === '1'
          ? scheduleService
          : null,
      scheduledTaskService:
        env.H5_SCHEDULE_ENABLED === '1'
          ? scheduledTaskService
          : null,
      intentDraftService:
        env.H5_INTENT_TRANSACTION_ENABLED === '1'
          ? intentDraftService
          : null,
      taskUnifiedService:
        env.H5_UNIFIED_TASKS_ENABLED === '1'
          ? taskUnifiedService
          : null,
      wechatScheduleLlmConfigService,
      wechatSubscribeMorningLlmConfigService,
      wechatScheduledTaskManageLlmConfigService,
      llmProviderService,
      chatIntentRouter,
      wechatIntentRouter,
      mysqlPool: pool,
      systemDisclosurePolicyService,
      sessionIntentClassifier: ({ text }) =>
        chatIntentRouter?.classifySessionAction({
          text,
        }),
      onPageGenerated: async ({
        userId,
        sessionId,
        artifacts = [],
      }) => {
        for (const artifact of artifacts) {
          const pageOwner =
            (await userAuth
              ?.getUserById(userId)
              .catch(() => null)) ?? {};
          const analyticsPayload = {
            eventName: 'page_generated',
            ownerId: userId,
            ownerSegment:
              resolveAnalyticsOwnerSegmentFn(
                pageOwner,
              ),
            ownerLabel: resolveAnalyticsOwnerLabelFn(
              pageOwner,
            ),
            planType: resolveAnalyticsPlanFn(pageOwner),
            generatedAt: new Date().toISOString(),
            pageId: artifact.relativePath,
            publicationId: sessionId,
            agentRunId: sessionId,
            channel: 'wechat_mp',
            url:
              artifact.url ||
              artifact.relativePath ||
              '/',
          };
          void sendMindSpaceAnalyticsEventFn({
            config: mindSpaceAnalyticsConfig,
            ...analyticsPayload,
          });
        }
      },
      applySessionLlmProvider: (sessionId, options = {}) =>
        tkmindProxy.applySessionLlmProvider(sessionId, options),
      refreshSessionSnapshot:
        sessionSnapshotService?.isEnabled()
          ? (sessionId, userId) =>
              sessionSnapshotService.refresh(
                sessionId,
                userId,
                async (pathname, init) => {
                  const target =
                    await tkmindProxy.resolveTarget(
                      sessionId,
                    );
                  return tkmindProxy.apiFetchTo(
                    target,
                    pathname,
                    init,
                  );
                },
              )
          : null,
      healthChannelStore: healthChannelStore ?? undefined,
      healthObservationStore,
      healthObservationService,
      healthDocumentStore,
      healthEventStore: healthEventStore ?? healthDataRuntime?.eventStore ?? null,
      env,
    });

  const notificationDispatcher =
    createNotificationDispatcherFn({
      sendWechatTextToUser: wechatMpService?.enabled
        ? (userId, text, options) =>
            wechatMpService.sendTextToUser(
              userId,
              text,
              options,
            )
        : null,
    });
  userAuth.setRechargeNotifier(
    async ({ userId, title, body, dedupeKey }) => {
      await notificationDispatcher.sendRechargeSuccess({
        userId,
        title,
        body,
        dedupeKey,
      });
    },
  );

  let scheduleReminderWorker = null;
  if (
    !isPassiveCanaryRuntime(env) &&
    env.H5_REMINDER_WORKER_ENABLED === '1' &&
    wechatMpService?.enabled &&
    scheduleService
  ) {
    scheduleReminderWorker =
      startScheduleReminderWorkerFn({
        scheduleService,
        notificationDispatcher,
      });
    logger.log('Schedule reminder worker enabled');
  }

  let scheduledTaskWorker = null;
  if (
    !isPassiveCanaryRuntime(env) &&
    isScheduledTaskWorkerEnabled(env) &&
    scheduledTaskService &&
    userAuth &&
    tkmindProxy
  ) {
    scheduledTaskWorker =
      startScheduledTaskWorkerFn({
        scheduledTaskService,
        scheduleService,
        userAuth,
        tkmindProxy,
        agentRunGateway,
        cursorExecutorPolicyService: wechatCursorExecutorPolicyService,
        sessionSnapshotService,
        notificationDispatcher,
        pool,
        h5Root,
        logger,
      });
    logger.log('Scheduled task worker enabled');
  } else if (
    !isPassiveCanaryRuntime(env)
    && scheduledTaskService
    && !isScheduledTaskWorkerEnabled(env)
  ) {
    logger.warn?.(
      'Scheduled task worker disabled: set H5_SCHEDULED_TASK_WORKER_ENABLED=1 or H5_REMINDER_WORKER_ENABLED=1',
    );
  }

  const wechatNewsMorningDraftService = createWechatNewsMorningDraftServiceFn(pool, {
    mpConfig: wechatMpConfig,
    h5Root,
    memindLibRoot: codeRoot || h5Root,
    env,
  });

  let wechatNewsMorningDraftWorker = null;
  if (
    !isPassiveCanaryRuntime(env)
    && wechatMpConfig?.enabled
    && userAuth
    && tkmindProxy
  ) {
    wechatNewsMorningDraftWorker = startWechatNewsMorningDraftWorkerFn({
      wechatNewsMorningDraftService,
      mpConfig: wechatMpConfig,
      userAuth,
      tkmindProxy,
      agentRunGateway,
      cursorExecutorPolicyService: wechatCursorExecutorPolicyService,
      sessionSnapshotService,
      pool,
      h5Root,
      env,
      logger,
    });
    if (wechatNewsMorningDraftWorker?.runOnce) {
      logger.log?.('WeChat news morning draft worker enabled');
    }
  }

  let healthBaselineWorker = null;
  if (!isPassiveCanaryRuntime(env) && healthDataRuntime) {
    const healthEventNotificationService = createHealthEventNotificationService({
      createUserNotification: scheduleService?.createUserNotification?.bind(scheduleService) ?? null,
      notificationDispatcher,
      logger,
    });
    healthBaselineWorker = startHealthBaselineWorkerFn({
      healthDataRuntime,
      eventNotificationService: healthEventNotificationService,
      pool,
      env,
      logger,
    });
    if (healthBaselineWorker?.runOnce) {
      logger.log?.('Health baseline job worker enabled');
    }
  }

  let subscriptionExpiryTimer = null;
  if (subscriptionService && !isPassiveCanaryRuntime(env)) {
    subscriptionExpiryTimer = setIntervalFn(
      async () => {
        try {
          const { renewed, failed } =
            await subscriptionService.processAutoRenewals();
          if (renewed > 0) {
            logger.log(
              `Auto-renewed ${renewed} subscription(s)`,
            );
          }
          if (failed > 0) {
            logger.log(
              `Auto-renew failed for ${failed} subscription(s) (balance insufficient)`,
            );
          }
          const expired =
            await subscriptionService.expireStaleSubscriptions();
          if (expired > 0) {
            logger.log(
              `Expired ${expired} stale subscription(s)`,
            );
          }
        } catch (error) {
          logger.warn(
            'Subscription expiry check failed:',
            error,
          );
        }
      },
      60 * 60 * 1000,
    );
    subscriptionExpiryTimer.unref?.();
  }

  const mindSpacePageEditSession =
    createPageEditSessionServiceFn({
      apiTarget,
      apiSecret,
      userAuth,
      sessionAccess,
      pageService: mindSpacePages,
      pageLiveEdit: mindSpacePageLiveEdit,
      llmProviderService,
    });
  if (wechatMpService?.enabled) {
    logger.log('WeChat MP webhook enabled');
  }
  logger.log(
    `User auth enabled (MySQL), workspace root: ${usersRoot}`,
  );

  return {
    wechatMpService,
    notificationDispatcher,
    scheduleReminderWorker,
    scheduledTaskWorker,
    wechatNewsMorningDraftService,
    wechatNewsMorningDraftWorker,
    healthBaselineWorker,
    subscriptionExpiryTimer,
    mindSpacePageEditSession,
  };
}