add

package.json

@@ -71,6 +71,7 @@
     "prepare": "husky",
     "prettier": "prettier -c --write \"**/**\"",
     "pull": "git pull",
+    "qstash": "pnpx @upstash/qstash-cli@latest dev",
     "reinstall": "rm -rf pnpm-lock.yaml && rm -rf node_modules && pnpm -r exec rm -rf node_modules && pnpm install",
     "reinstall:desktop": "rm -rf pnpm-lock.yaml && rm -rf node_modules && pnpm -r exec rm -rf node_modules && pnpm install --node-linker=hoisted",
     "release": "semantic-release",
@@ -84,8 +85,9 @@
     "test:e2e": "pnpm --filter @lobechat/e2e-tests test",
     "test:e2e:smoke": "pnpm --filter @lobechat/e2e-tests test:smoke",
     "test:update": "vitest -u",
+    "tunnel:cloudflare": "cloudflared tunnel --url http://localhost:3010",
+    "tunnel:ngrok": "ngrok http http://localhost:3011",
     "type-check": "tsgo --noEmit",
-    "webhook:ngrok": "ngrok http http://localhost:3011",
     "workflow:cdn": "tsx ./scripts/cdnWorkflow/index.ts",
     "workflow:changelog": "tsx ./scripts/changelogWorkflow/index.ts",
     "workflow:countCharters": "tsx scripts/countEnWord.ts",
@@ -189,6 +191,7 @@
     "@trpc/next": "^11.7.1",
     "@trpc/react-query": "^11.7.1",
     "@trpc/server": "^11.7.1",
+    "@upstash/qstash": "^2.8.2",
     "@vercel/analytics": "^1.5.0",
     "@vercel/edge-config": "^1.4.3",
     "@vercel/functions": "^3.3.2",
@@ -221,6 +224,7 @@
     "i18next-browser-languagedetector": "^8.2.0",
     "i18next-resources-to-backend": "^1.2.1",
     "immer": "^10.2.0",
+    "ioredis": "^5.7.0",
     "jose": "^5.10.0",
     "js-sha256": "^0.11.1",
     "jsonl-parse-stringify": "^1.0.3",

packages/database/src/core/dbForTest.ts

@@ -7,7 +7,7 @@ import { join } from 'node:path';
 import { Pool as NodePool } from 'pg';
 import ws from 'ws';
 
-import { serverDBEnv } from '@/config/db';
+import { serverDBEnv } from '@/envs/db';
 
 import * as schema from '../schemas';
 

packages/database/src/core/web-server.ts

@@ -4,7 +4,7 @@ import { drizzle as nodeDrizzle } from 'drizzle-orm/node-postgres';
 import { Pool as NodePool } from 'pg';
 import ws from 'ws';
 
-import { serverDBEnv } from '@/config/db';
+import { serverDBEnv } from '@/envs/db';
 
 import * as schema from '../schemas';
 import { LobeChatDatabase } from '../type';

packages/utils/src/server/index.ts

@@ -3,4 +3,5 @@ export * from './correctOIDCUrl';
 export * from './geo';
 export * from './response';
 export * from './responsive';
+export * from './sse';
 export * from './xor';

packages/utils/src/server/sse.ts

@@ -0,0 +1,151 @@
+/**
+ * SSE (Server-Sent Events) utilities for agent streaming
+ */
+
+export interface SSEEvent {
+  data: any;
+  event?: string;
+  id?: string;
+  retry?: number;
+}
+
+/**
+ * Formats data into SSE format with id/event/data structure
+ * @param event - The SSE event configuration
+ * @returns Formatted SSE string
+ */
+export function formatSSEEvent({ id, event, data, retry }: SSEEvent): string {
+  const lines: string[] = [];
+
+  if (id !== undefined) {
+    lines.push(`id: ${id}`);
+  }
+
+  if (event !== undefined) {
+    lines.push(`event: ${event}`);
+  }
+
+  if (retry !== undefined) {
+    lines.push(`retry: ${retry}`);
+  }
+
+  // Handle data serialization
+  const dataString = typeof data === 'string' ? data : JSON.stringify(data);
+
+  // Split multi-line data and prefix each line with "data: "
+  const dataLines = dataString.split('\n');
+  dataLines.forEach((line) => {
+    lines.push(`data: ${line}`);
+  });
+
+  // End with double newline
+  lines.push('', '');
+
+  return lines.join('\n');
+}
+
+/**
+ * Creates a utility for enqueueing SSE events to a ReadableStream controller
+ * @param controller - The ReadableStreamDefaultController
+ * @returns Helper function for sending SSE events
+ */
+export function createSSEWriter(controller: ReadableStreamDefaultController<string>) {
+  return {
+    /**
+     * Send a connection event
+     */
+    writeConnection(sessionId: string, lastEventId: string, timestamp: number = Date.now()) {
+      this.writeEvent({
+        data: {
+          lastEventId,
+          sessionId,
+          timestamp,
+          type: 'connected',
+        },
+        event: 'connected',
+        id: `conn_${timestamp}`,
+      });
+    },
+
+    /**
+     * Send an error event
+     */
+    writeError(error: any, sessionId: string, phase?: string, timestamp: number = Date.now()) {
+      this.writeEvent({
+        data: {
+          data: {
+            error: error.message || String(error),
+            phase: phase || 'unknown',
+            ...(error.stack && { stack: error.stack }),
+          },
+          sessionId,
+          timestamp,
+          type: 'error',
+        },
+        event: 'error',
+        id: `error_${timestamp}`,
+      });
+    },
+
+    /**
+     * Send an SSE event
+     */
+    writeEvent(event: SSEEvent) {
+      controller.enqueue(formatSSEEvent(event));
+    },
+
+    /**
+     * Send a heartbeat/keep-alive event
+     */
+    writeHeartbeat(timestamp: number = Date.now()) {
+      this.writeEvent({
+        data: {
+          timestamp,
+          type: 'heartbeat',
+        },
+        event: 'heartbeat',
+        id: `heartbeat_${timestamp}`,
+      });
+    },
+
+    /**
+     * Send a stream event (for historical or real-time events)
+     */
+    writeStreamEvent(eventData: any, eventId?: string) {
+      this.writeEvent({
+        data: eventData,
+        event: eventData.type || 'stream',
+        id: eventId || `event_${Date.now()}`,
+      });
+    },
+  };
+}
+
+/**
+ * Agent stream event types
+ */
+export type AgentStreamEventType =
+  | 'connected'
+  | 'stream'
+  | 'error'
+  | 'heartbeat'
+  | 'stream_start'
+  | 'stream_chunk'
+  | 'stream_end'
+  | 'stream_error';
+
+/**
+ * Creates SSE headers for agent streaming
+ */
+// eslint-disable-next-line no-undef
+export function createSSEHeaders(): HeadersInit {
+  return {
+    'Access-Control-Allow-Headers': 'Cache-Control, Last-Event-ID',
+    'Access-Control-Allow-Methods': 'GET',
+    'Access-Control-Allow-Origin': '*',
+    'Cache-Control': 'no-cache, no-transform',
+    'Connection': 'keep-alive',
+    'Content-Type': 'text/event-stream',
+    'X-Accel-Buffering': 'no',
+  };
+}

src/app/(backend)/api/auth/adapter/route.ts

@@ -1,8 +1,8 @@
 import debug from 'debug';
 import { NextRequest, NextResponse } from 'next/server';
 
-import { serverDBEnv } from '@/config/db';
 import { serverDB } from '@/database/server';
+import { serverDBEnv } from '@/envs/db';
 import { dateKeys } from '@/libs/next-auth/adapter';
 import { NextAuthUserService } from '@/server/services/nextAuthUser';
 

src/app/(backend)/webapi/create-image/comfyui/route.ts

@@ -1,7 +1,7 @@
 import { NextResponse } from 'next/server';
 
 import { checkAuth } from '@/app/(backend)/middleware/auth';
-import { getServerDBConfig } from '@/config/db';
+import { getServerDBConfig } from '@/envs/db';
 import { createCallerFactory } from '@/libs/trpc/lambda';
 import { lambdaRouter } from '@/server/routers/lambda';
 

src/envs/db.ts

@@ -10,6 +10,9 @@ export const getServerDBConfig = () => {
 
       KEY_VAULTS_SECRET: process.env.KEY_VAULTS_SECRET,
 
+      QSTASH_TOKEN: process.env.QSTASH_TOKEN,
+      REDIS_URL: process.env.REDIS_URL,
+
       REMOVE_GLOBAL_FILE: process.env.DISABLE_REMOVE_GLOBAL_FILE !== '0',
     },
     server: {
@@ -19,6 +22,9 @@ export const getServerDBConfig = () => {
 
       KEY_VAULTS_SECRET: z.string().optional(),
 
+      QSTASH_TOKEN: z.string().optional(),
+      REDIS_URL: z.string().optional(),
+
       REMOVE_GLOBAL_FILE: z.boolean().optional(),
     },
   });

src/libs/next-auth/adapter/index.ts

@@ -8,8 +8,8 @@ import debug from 'debug';
 import { Adapter, AdapterAccount } from 'next-auth/adapters';
 import urlJoin from 'url-join';
 
-import { serverDBEnv } from '@/config/db';
 import { appEnv } from '@/envs/app';
+import { serverDBEnv } from '@/envs/db';
 
 const log = debug('lobe-next-auth:adapter');
 

src/libs/oidc-provider/provider.ts

@@ -4,10 +4,10 @@ import debug from 'debug';
 import Provider, { Configuration, KoaContextWithOIDC, errors } from 'oidc-provider';
 import urlJoin from 'url-join';
 
-import { serverDBEnv } from '@/config/db';
 import { enableClerk } from '@/const/auth';
 import { UserModel } from '@/database/models/user';
 import { appEnv } from '@/envs/app';
+import { serverDBEnv } from '@/envs/db';
 import { getJWKS } from '@/libs/oidc-provider/jwt';
 import { normalizeLocale } from '@/locales/resources';
 

src/libs/redis/client.ts

@@ -0,0 +1,64 @@
+import debug from 'debug';
+import Redis from 'ioredis';
+
+import { getRedisConnectionDescription, getRedisUrl } from './config';
+
+const log = debug('redis:client');
+
+/**
+ * 创建 Redis 客户端实例
+ */
+export const createRedisClient = (url?: string): Redis | null => {
+  const redisUrl = url || getRedisUrl();
+
+  if (!redisUrl) {
+    console.warn('[Redis Client] No Redis URL available. Redis features are disabled.');
+    return null;
+  }
+
+  const client = new Redis(redisUrl, {
+    maxRetriesPerRequest: 3,
+  });
+
+  client.on('connect', () => {
+    log(`Connected to Redis: ${getRedisConnectionDescription(redisUrl)}`);
+  });
+
+  client.on('error', (error) => {
+    console.error('[Redis Client] Redis connection error:', error);
+  });
+
+  client.on('close', () => {
+    log('Redis connection closed');
+  });
+
+  return client;
+};
+
+/**
+ * 全局 Redis 客户端实例（单例模式）
+ */
+let globalRedisClient: Redis | null = null;
+let redisInitialized = false;
+
+/**
+ * 获取全局 Redis 客户端实例
+ */
+export function getRedisClient(): Redis | null {
+  if (!redisInitialized) {
+    globalRedisClient = createRedisClient();
+    redisInitialized = true;
+  }
+  return globalRedisClient;
+}
+
+/**
+ * 关闭全局 Redis 客户端连接
+ */
+export async function closeRedisClient(): Promise<void> {
+  if (globalRedisClient) {
+    await globalRedisClient.quit();
+    globalRedisClient = null;
+    redisInitialized = false;
+  }
+}

src/libs/redis/config.ts

@@ -0,0 +1,40 @@
+/**
+ * Get Redis URL from environment variables
+ */
+export const getRedisUrl = (): string | undefined => {
+  const redisUrl = process.env.REDIS_URL;
+
+  if (!redisUrl) {
+    return undefined;
+  }
+
+  return redisUrl;
+};
+
+/**
+ * Validate if Redis URL is valid
+ */
+export const validateRedisUrl = (url: string): boolean => {
+  try {
+    new URL(url);
+    return true;
+  } catch {
+    console.error('[Redis Config] Invalid REDIS_URL format:', url);
+    return false;
+  }
+};
+
+/**
+ * Get Redis connection description string for logging (hide password)
+ */
+export const getRedisConnectionDescription = (url: string): string => {
+  try {
+    const urlObj = new URL(url);
+    if (urlObj.password) {
+      urlObj.password = '***';
+    }
+    return urlObj.toString();
+  } catch {
+    return 'Invalid URL';
+  }
+};

src/libs/redis/index.ts

@@ -0,0 +1,2 @@
+export { closeRedisClient,createRedisClient, getRedisClient } from './client';
+export { getRedisConnectionDescription,getRedisUrl, validateRedisUrl } from './config';

src/libs/trpc/async/asyncAuth.ts

@@ -2,8 +2,8 @@ import { LobeChatDatabase } from '@lobechat/database';
 import { TRPCError } from '@trpc/server';
 import debug from 'debug';
 
-import { serverDBEnv } from '@/config/db';
 import { UserModel } from '@/database/models/user';
+import { serverDBEnv } from '@/envs/db';
 
 import { asyncTrpc } from './init';
 

src/server/modules/KeyVaultsEncrypt/index.ts

@@ -1,4 +1,4 @@
-import { getServerDBConfig } from '@/config/db';
+import { getServerDBConfig } from '@/envs/db';
 import { UserKeyVaults } from '@/types/user/settings';
 
 interface DecryptionResult {

src/server/routers/async/caller.ts

@@ -3,10 +3,10 @@ import { createTRPCClient, httpLink } from '@trpc/client';
 import superjson from 'superjson';
 import urlJoin from 'url-join';
 
-import { serverDBEnv } from '@/config/db';
 import { LOBE_CHAT_AUTH_HEADER } from '@/const/auth';
 import { isDesktop } from '@/const/version';
 import { appEnv } from '@/envs/app';
+import { serverDBEnv } from '@/envs/db';
 import { createAsyncCallerFactory } from '@/libs/trpc/async';
 import { createAsyncContextInner } from '@/libs/trpc/async/context';
 import { KeyVaultsGateKeeper } from '@/server/modules/KeyVaultsEncrypt';

src/server/routers/async/file.ts

@@ -3,13 +3,13 @@ import { chunk } from 'lodash-es';
 import pMap from 'p-map';
 import { z } from 'zod';
 
-import { serverDBEnv } from '@/config/db';
 import { DEFAULT_FILE_EMBEDDING_MODEL_ITEM } from '@/const/settings/knowledge';
 import { ASYNC_TASK_TIMEOUT, AsyncTaskModel } from '@/database/models/asyncTask';
 import { ChunkModel } from '@/database/models/chunk';
 import { EmbeddingModel } from '@/database/models/embedding';
 import { FileModel } from '@/database/models/file';
 import { NewChunkItem, NewEmbeddingsItem } from '@/database/schemas';
+import { serverDBEnv } from '@/envs/db';
 import { fileEnv } from '@/envs/file';
 import { asyncAuthedProcedure, asyncRouter as router } from '@/libs/trpc/async';
 import { getServerDefaultFilesConfig } from '@/server/globalConfig';

src/server/routers/lambda/file.ts

@@ -1,12 +1,12 @@
 import { TRPCError } from '@trpc/server';
 import { z } from 'zod';
 
-import { serverDBEnv } from '@/config/db';
 import { AsyncTaskModel } from '@/database/models/asyncTask';
 import { ChunkModel } from '@/database/models/chunk';
 import { DocumentModel } from '@/database/models/document';
 import { FileModel } from '@/database/models/file';
 import { KnowledgeRepo } from '@/database/repositories/knowledge';
+import { serverDBEnv } from '@/envs/db';
 import { authedProcedure, router } from '@/libs/trpc/lambda';
 import { serverDatabase } from '@/libs/trpc/lambda/middleware';
 import { FileService } from '@/server/services/file';

src/server/services/file/index.ts

@@ -3,9 +3,9 @@ import { inferContentTypeFromImageUrl, nanoid, uuid } from '@lobechat/utils';
 import { TRPCError } from '@trpc/server';
 import { sha256 } from 'js-sha256';
 
-import { serverDBEnv } from '@/config/db';
 import { FileModel } from '@/database/models/file';
 import { FileItem } from '@/database/schemas';
+import { serverDBEnv } from '@/envs/db';
 import { TempFileManager } from '@/server/utils/tempFileManager';
 
 import { FileServiceImpl, createFileServiceModule } from './impls';

src/store/chat/slices/aiChat/actions/conversationLifecycle.ts

@@ -173,7 +173,7 @@ export const conversationLifecycle: StateCreator<
           newTopic: shouldCreateNewTopic
             ? {
                 topicMessageIds: messages.map((m) => m.id),
-                title: message.slice(0, 10) || t('defaultTitle', { ns: 'topic' }),
+                title: message.slice(0, 20) || t('defaultTitle', { ns: 'topic' }),
               }
             : undefined,
           sessionId: activeId === INBOX_SESSION_ID ? undefined : activeId,