curriculum-project-hub/hub/src/hub.ts

import Fastify from "fastify";
import { registerAdminPlugin } from "./admin/plugin.js";
import { prisma } from "./db.js";
import { createLarkClient, sendText, startFeishuListenerWithClient, type FeishuRuntime } from "./feishu/client.js";
import { archiveFeishuBindingForLifecycleEvent } from "./feishu/bindingLifecycle.js";
import { makeTriggerHandler } from "./feishu/trigger.js";
import { removeAbandonedMessageResourceStages } from "./feishu/resourceStaging.js";
import { triggerQueue } from "./feishu/triggerQueue.js";
import { LocalSecretEnvelope, loadLocalSecretKeyring } from "./security/secretEnvelope.js";
import { createDatabaseRuntimeSettings } from "./settings/runtime.js";
import { verifyStoredProviderEnvelopes } from "./connections/providerConnections.js";
import { openProviderProxyLease } from "./connections/providerProxy.js";
import { verifyStoredFeishuApplicationEnvelopes } from "./connections/feishuApplicationConnections.js";
import { resolveActiveFeishuApplication } from "./connections/feishuApplicationConnections.js";
import { readServerBinding } from "./settings/server.js";
import { readSiloOrganizationId, requireSiloOrganization } from "./deployment/silo.js";
import { isSiloHttpRateLimitExempt, SiloFixedWindowRateLimiter } from "./deployment/siloRateLimit.js";

function requireEnv(name: string): string {
  const value = process.env[name];
  if (value === undefined || value === "") {
    throw new Error(`missing required env: ${name}`);
  }
  return value;
}

function booleanEnv(name: string, fallback: boolean): boolean {
  const raw = process.env[name];
  if (raw === undefined || raw === "") return fallback;
  return !["0", "false", "no", "off"].includes(raw.trim().toLowerCase());
}

/** Start the Hub after every import and runtime dependency has loaded. */
export async function startHub(): Promise<void> {
  requireEnv("DATABASE_URL");
  const projectWorkspaceRoot = requireEnv("HUB_PROJECT_WORKSPACE_ROOT");
  const httpBodyLimit = positiveIntegerEnv("HUB_HTTP_BODY_LIMIT_BYTES");
  const maxFilesPerMessage = positiveIntegerEnv("HUB_MAX_FILES_PER_MESSAGE");
  const maxBytesPerFile = positiveIntegerEnv("HUB_MAX_FILE_BYTES");
  const httpRequestsPerMinute = positiveIntegerEnv("HUB_HTTP_REQUESTS_PER_MINUTE");
  const feishuEventsPerMinute = positiveIntegerEnv("HUB_FEISHU_EVENTS_PER_MINUTE");
  const app = Fastify({ logger: true, bodyLimit: httpBodyLimit });
  const requestLimiter = new SiloFixedWindowRateLimiter(httpRequestsPerMinute, 60_000);
  app.addHook("onRequest", async (request, reply) => {
    // Static SPA assets / admin HTML shell / healthz do not count toward the
    // silo requestRate budget (a full admin load can pull dozens of chunks).
    if (isSiloHttpRateLimitExempt(request.url)) return;
    const decision = requestLimiter.consume();
    if (!decision.allowed) {
      await reply
        .header("Retry-After", String(decision.retryAfterSeconds))
        .status(429)
        .send({ error: { code: "rate_limited", message: "Silo request rate exceeded" } });
    }
  });

  const abandonedStages = await removeAbandonedMessageResourceStages(projectWorkspaceRoot);
  app.log.info({ removed: abandonedStages }, "startup: removed abandoned Feishu resource stages");

  const secretEnvelope = new LocalSecretEnvelope(await loadLocalSecretKeyring());
  const verifiedProviderEnvelopes = await verifyStoredProviderEnvelopes(prisma, secretEnvelope);
  const verifiedFeishuEnvelopes = await verifyStoredFeishuApplicationEnvelopes(prisma, secretEnvelope);
  app.log.info(
    { provider: verifiedProviderEnvelopes, feishu: verifiedFeishuEnvelopes },
    "startup: authenticated stored connection envelopes",
  );
  const runtimeSettings = createDatabaseRuntimeSettings(
    prisma,
    secretEnvelope,
    process.env,
    (credential, context) => openProviderProxyLease(credential, {
      onDiagnostic: (diagnostic) => {
        app.log.error({
          runId: context.runId,
          projectId: context.projectId,
          providerId: context.providerId,
          errorCode: diagnostic.code,
          failureCategory: diagnostic.category,
        }, "provider proxy diagnostic");
      },
    }),
  );

  const siloOrganizationId = readSiloOrganizationId();
  const siloOrganization = await requireSiloOrganization(prisma, siloOrganizationId);
  const activeProvider = await prisma.organizationProviderConnection.findUnique({
    where: {
      organizationId_providerId: {
        organizationId: siloOrganization.id,
        providerId: "openrouter",
      },
    },
    select: {
      status: true,
      activeSecretVersion: { select: { connectionId: true, retiredAt: true } },
      id: true,
    },
  });
  if (activeProvider?.status !== "ACTIVE" || activeProvider.activeSecretVersion === null ||
      activeProvider.activeSecretVersion.connectionId !== activeProvider.id ||
      activeProvider.activeSecretVersion.retiredAt !== null) {
    throw new Error(`Silo Organization ${siloOrganization.id} has no ACTIVE openrouter Provider Connection`);
  }
  const feishuApplication = await resolveActiveFeishuApplication(
    prisma,
    secretEnvelope,
    { organizationId: siloOrganization.id },
  );
  app.log.info(
    {
      organizationId: siloOrganization.id,
      organizationSlug: siloOrganization.slug,
      feishuConnectionId: feishuApplication.connectionId,
    },
    "startup: proved Silo Organization and resolved Feishu Application",
  );
  const sessionSecret = requireEnv("HUB_SESSION_SECRET");
  const publicBaseUrl = process.env["HUB_PUBLIC_BASE_URL"] ?? "http://127.0.0.1:8788";
  const bind = readServerBinding();

  // Startup reset: clear stale locks + mark dead runs as FAILED. Capture the
  // killed runs first so we can tell their Feishu chats after the listener is up.
  const interruptedRuns = await prisma.agentRun.findMany({
    where: { status: "ACTIVE" },
    select: { id: true, projectId: true },
  });
  await prisma.projectAgentLock.deleteMany({});
  if (interruptedRuns.length > 0) {
    await prisma.agentRun.updateMany({
      where: { id: { in: interruptedRuns.map((run) => run.id) } },
      data: { status: "FAILED", error: "process restart", finishedAt: new Date() },
    });
  }
  app.log.info({ killedRuns: interruptedRuns.length }, "startup: cleared stale locks + dead runs");

  let feishuRuntime: FeishuRuntime | undefined;
  app.get("/api/healthz", async (_request, reply) => {
    const feishuReady = feishuRuntime?.isListenerReady?.() ?? !booleanEnv("HUB_FEISHU_LISTENER_ENABLED", true);
    if (!feishuReady) return reply.status(503).send({ ok: false, feishuReady, ts: Date.now() });
    return { ok: true, feishuReady, ts: Date.now() };
  });

  await registerAdminPlugin(app, {
    prisma,
    sessionSecret,
    publicBaseUrl,
    feishuAppId: feishuApplication.appId,
    feishuAppSecret: feishuApplication.appSecret,
    projectWorkspaceRoot,
    secretEnvelope,
  });

  const feishuListenerEnabled = booleanEnv("HUB_FEISHU_LISTENER_ENABLED", true);
  if (feishuListenerEnabled) {
    const feishuConfig = {
      appId: feishuApplication.appId,
      appSecret: feishuApplication.appSecret,
      botOpenId: feishuApplication.botOpenId,
    };
    const larkClient = createLarkClient(feishuConfig);
    const triggerQueuePurgeTimer = setInterval(() => {
      const removed = triggerQueue.purgeExpired();
      if (removed > 0) {
        app.log.info({ removed }, "feishu trigger queue: purged expired triggers");
      }
    }, 60_000);
    triggerQueuePurgeTimer.unref();
    const trigger = makeTriggerHandler({
      prisma,
      settings: runtimeSettings,
      logger: app.log,
      secretEnvelope,
      projectWorkspaceRoot,
      publicBaseUrl,
      siloOrganizationId: siloOrganization.id,
      rejectWhenBusy: true,
      resourceLimits: { maxFilesPerMessage, maxBytesPerFile },
      allowLegacyFeishuIdentity: false,
      maxFeishuEventsPerMinute: feishuEventsPerMinute,
    });
    const runtime = await startFeishuListenerWithClient(
      feishuConfig,
      larkClient,
      app.log,
      trigger,
      trigger.onCardAction,
      () => {
        process.exitCode = 1;
        void app.close().catch((error) => app.log.error({ err: error }, "Hub close after Feishu failure failed"));
      },
      async (event) => {
        const result = await archiveFeishuBindingForLifecycleEvent(prisma, event);
        app.log.info({ ...event, archived: result.archived, projectId: result.projectId }, "feishu binding lifecycle event handled");
      },
    );
    feishuRuntime = runtime;
    await notifyBoundChatsOfInterruptedRuns(runtime, prisma, interruptedRuns, app.log);
  } else {
    app.log.info("feishu listener disabled by HUB_FEISHU_LISTENER_ENABLED");
  }
  const address = await app.listen(bind);
  app.log.info({ address }, "hub listening");
}

async function notifyBoundChatsOfInterruptedRuns(
  rt: FeishuRuntime,
  db: typeof prisma,
  interruptedRuns: ReadonlyArray<{ readonly id: string; readonly projectId: string }>,
  logger: { info: (obj: unknown, msg?: string) => void; warn: (obj: unknown, msg?: string) => void },
): Promise<void> {
  if (interruptedRuns.length === 0) return;
  const byProject = new Map<string, string[]>();
  for (const run of interruptedRuns) {
    const list = byProject.get(run.projectId) ?? [];
    list.push(run.id);
    byProject.set(run.projectId, list);
  }
  for (const [projectId, runIds] of byProject) {
    const binding = await db.projectGroupBinding.findFirst({
      where: { projectId, archivedAt: null },
      select: { chatId: true },
    });
    if (binding === null) continue;
    const n = runIds.length;
    const text =
      n === 1
        ? `\u26A0\uFE0F \u670D\u52A1\u521A\u521A\u91CD\u542F\uFF0C\u4E0A\u4E00\u4E2A\u6B63\u5728\u8FDB\u884C\u7684\u4EFB\u52A1\u88AB\u4E2D\u65AD\u4E86\u3002\u8BF7\u91CD\u65B0\u53D1\u8D77\u8BF7\u6C42\uFF08run: ${runIds[0]}\uFF09\u3002`
        : `\u26A0\uFE0F \u670D\u52A1\u521A\u521A\u91CD\u542F\uFF0C${n} \u4E2A\u6B63\u5728\u8FDB\u884C\u7684\u4EFB\u52A1\u88AB\u4E2D\u65AD\u4E86\u3002\u8BF7\u91CD\u65B0\u53D1\u8D77\u8BF7\u6C42\u3002`;
    try {
      await sendText(rt, binding.chatId, text);
      logger.info({ projectId, chatId: binding.chatId, runIds }, "startup: notified chat of interrupted runs");
    } catch (error) {
      logger.warn({ projectId, chatId: binding.chatId, err: String(error) }, "startup: failed to notify chat of interrupted runs");
    }
  }
}

function positiveIntegerEnv(name: string): number {
  const raw = requireEnv(name);
  const value = Number(raw);
  if (!Number.isSafeInteger(value) || value <= 0) throw new Error(`${name} must be a positive safe integer`);
  return value;
}