feat(server,stage-ui): sync message to server (#990)

apps/server/drizzle/0006_gray_stardust.sql

@@ -0,0 +1,6 @@
+ALTER TABLE "chat_members" ALTER COLUMN "user_id" DROP NOT NULL;--> statement-breakpoint
+ALTER TABLE "chat_members" ADD COLUMN "member_type" text NOT NULL;--> statement-breakpoint
+ALTER TABLE "chat_members" ADD COLUMN "character_id" text;--> statement-breakpoint
+ALTER TABLE "chats" ADD COLUMN "title" text;--> statement-breakpoint
+ALTER TABLE "messages" ADD COLUMN "role" text NOT NULL;--> statement-breakpoint
+ALTER TABLE "chat_members" ADD CONSTRAINT "chat_members_character_id_characters_id_fk" FOREIGN KEY ("character_id") REFERENCES "public"."characters"("id") ON DELETE cascade ON UPDATE no action;

apps/server/drizzle/meta/_journal.json

@@ -43,6 +43,13 @@
       "when": 1767792829241,
       "tag": "0005_true_namor",
       "breakpoints": true
+    },
+    {
+      "idx": 6,
+      "version": "7",
+      "when": 1769186900274,
+      "tag": "0006_gray_stardust",
+      "breakpoints": true
     }
   ]
 }

apps/server/src/api/chats.schema.ts

@@ -0,0 +1,45 @@
+import { array, literal, number, object, optional, string, union } from 'valibot'
+
+const ChatTypeSchema = union([
+  literal('private'),
+  literal('bot'),
+  literal('group'),
+  literal('channel'),
+])
+
+const ChatMemberTypeSchema = union([
+  literal('user'),
+  literal('character'),
+  literal('bot'),
+])
+
+const ChatMessageRoleSchema = union([
+  literal('system'),
+  literal('user'),
+  literal('assistant'),
+  literal('tool'),
+  literal('error'),
+])
+
+export const ChatSyncMessageSchema = object({
+  id: string(),
+  role: ChatMessageRoleSchema,
+  content: string(),
+  createdAt: optional(number()),
+})
+
+export const ChatSyncSchema = object({
+  chat: object({
+    id: string(),
+    type: optional(ChatTypeSchema),
+    title: optional(string()),
+    createdAt: optional(number()),
+    updatedAt: optional(number()),
+  }),
+  members: optional(array(object({
+    type: ChatMemberTypeSchema,
+    userId: optional(string()),
+    characterId: optional(string()),
+  }))),
+  messages: array(ChatSyncMessageSchema),
+})

apps/server/src/app.ts

@@ -11,9 +11,11 @@ import { createLoggLogger, injeca } from 'injeca'
 
 import { sessionMiddleware } from './middlewares/auth'
 import { createCharacterRoutes } from './routes/characters'
+import { createChatRoutes } from './routes/chats'
 import { createProviderRoutes } from './routes/providers'
 import { createAuth } from './services/auth'
 import { createCharacterService } from './services/characters'
+import { createChatService } from './services/chats'
 import { createDrizzle } from './services/db'
 import { parsedEnv } from './services/env'
 import { createProviderService } from './services/providers'
@@ -24,15 +26,17 @@ import * as schema from './schemas'
 
 type AuthService = ReturnType<typeof createAuth>
 type CharacterService = ReturnType<typeof createCharacterService>
+type ChatService = ReturnType<typeof createChatService>
 type ProviderService = ReturnType<typeof createProviderService>
 
 interface AppDeps {
   auth: AuthService
   characterService: CharacterService
+  chatService: ChatService
   providerService: ProviderService
 }
 
-function buildApp({ auth, characterService, providerService }: AppDeps) {
+function buildApp({ auth, characterService, chatService, providerService }: AppDeps) {
   const logger = useLogger('app').useGlobalConfig()
 
   return new Hono<HonoEnv>()
@@ -77,6 +81,11 @@ function buildApp({ auth, characterService, providerService }: AppDeps) {
      * Provider routes are handled by the provider service.
      */
     .route('/api/providers', createProviderRoutes(providerService))
+
+    /**
+     * Chat routes are handled by the chat service.
+     */
+    .route('/api/chats', createChatRoutes(chatService))
 }
 
 export type AppType = ReturnType<typeof buildApp>
@@ -114,11 +123,17 @@ async function createApp() {
     build: ({ dependsOn }) => createProviderService(dependsOn.db),
   })
 
+  const chatService = injeca.provide('services:chats', {
+    dependsOn: { db },
+    build: ({ dependsOn }) => createChatService(dependsOn.db),
+  })
+
   await injeca.start()
-  const resolved = await injeca.resolve({ auth, characterService, providerService })
+  const resolved = await injeca.resolve({ auth, characterService, chatService, providerService })
   const app = buildApp({
     auth: resolved.auth,
     characterService: resolved.characterService,
+    chatService: resolved.chatService,
     providerService: resolved.providerService,
   })
 

apps/server/src/routes/chats.ts

@@ -0,0 +1,26 @@
+import type { ChatService } from '../services/chats'
+import type { HonoEnv } from '../types/hono'
+
+import { Hono } from 'hono'
+import { safeParse } from 'valibot'
+
+import { ChatSyncSchema } from '../api/chats.schema'
+import { authGuard } from '../middlewares/auth'
+import { createBadRequestError } from '../utils/error'
+
+export function createChatRoutes(chatService: ChatService) {
+  return new Hono<HonoEnv>()
+    .use('*', authGuard)
+    .post('/sync', async (c) => {
+      const user = c.get('user')!
+
+      const body = await c.req.json()
+      const result = safeParse(ChatSyncSchema, body)
+
+      if (!result.success)
+        throw createBadRequestError('Invalid Request', 'INVALID_REQUEST', result.issues)
+
+      const synced = await chatService.syncChat(user.id, result.output)
+      return c.json(synced)
+    })
+}

apps/server/src/schemas/chats.ts

@@ -1,7 +1,10 @@
+import type { InferInsertModel, InferSelectModel } from 'drizzle-orm'
+
 import { integer, pgTable, text, timestamp } from 'drizzle-orm/pg-core'
 
 import { nanoid } from '../utils/id'
 import { user } from './accounts'
+import { character } from './characters'
 
 export const media = pgTable(
   'media',
@@ -37,6 +40,7 @@ export const stickerPacks = pgTable(
 )
 
 type ChatType = 'private' | 'bot' | 'group' | 'channel'
+type ChatMemberType = 'user' | 'character' | 'bot'
 
 export const chats = pgTable(
   'chats',
@@ -44,6 +48,7 @@ export const chats = pgTable(
     id: text('id').primaryKey().$defaultFn(() => nanoid()),
 
     type: text('type').notNull().$type<ChatType>(),
+    title: text('title'),
 
     createdAt: timestamp('created_at').defaultNow().notNull(),
     updatedAt: timestamp('updated_at').defaultNow().notNull(),
@@ -51,12 +56,17 @@ export const chats = pgTable(
   },
 )
 
+export type Chat = InferSelectModel<typeof chats>
+export type NewChat = InferInsertModel<typeof chats>
+
 export const chatMembers = pgTable(
   'chat_members',
   {
     id: text('id').primaryKey().$defaultFn(() => nanoid()),
     chatId: text('chat_id').notNull().references(() => chats.id, { onDelete: 'cascade' }),
-    userId: text('user_id').notNull().references(() => user.id, { onDelete: 'cascade' }),
+    memberType: text('member_type').notNull().$type<ChatMemberType>(),
+    userId: text('user_id').references(() => user.id, { onDelete: 'cascade' }),
+    characterId: text('character_id').references(() => character.id, { onDelete: 'cascade' }),
   },
 )
 
@@ -67,6 +77,7 @@ export const messages = pgTable(
 
     chatId: text('chat_id').notNull().references(() => chats.id, { onDelete: 'cascade' }),
     senderId: text('sender_id').notNull(),
+    role: text('role').notNull(),
 
     content: text('content').notNull(),
     mediaIds: text('media_ids').array().notNull(),
@@ -80,3 +91,6 @@ export const messages = pgTable(
     deletedAt: timestamp('deleted_at'),
   },
 )
+
+export type Message = InferSelectModel<typeof messages>
+export type NewMessage = InferInsertModel<typeof messages>

apps/server/src/services/chats.ts

@@ -0,0 +1,172 @@
+import type * as fullSchema from '../schemas'
+import type { Database } from './db'
+
+import { and, eq } from 'drizzle-orm'
+
+import { createConflictError, createForbiddenError } from '../utils/error'
+
+import * as schema from '../schemas/chats'
+
+type ChatType = 'private' | 'bot' | 'group' | 'channel'
+type MessageRole = 'system' | 'user' | 'assistant' | 'tool' | 'error'
+type ChatMemberType = 'user' | 'character' | 'bot'
+
+interface SyncChatMessagePayload {
+  id: string
+  role: MessageRole
+  content: string
+  createdAt?: number
+}
+
+interface SyncChatMemberPayload {
+  type: ChatMemberType
+  userId?: string
+  characterId?: string
+}
+
+interface SyncChatPayload {
+  chat: {
+    id: string
+    type?: ChatType
+    title?: string
+    createdAt?: number
+    updatedAt?: number
+  }
+  members?: SyncChatMemberPayload[]
+  messages: SyncChatMessagePayload[]
+}
+
+function resolveSenderId(role: MessageRole, userId: string, characterId?: string) {
+  if (role === 'user')
+    return userId
+  return characterId ?? role
+}
+
+function pickCharacterId(members: SyncChatMemberPayload[] | undefined) {
+  return members?.find(member => member.type === 'character' && member.characterId)?.characterId
+}
+
+export function createChatService(db: Database<typeof fullSchema>) {
+  return {
+    async syncChat(userId: string, payload: SyncChatPayload) {
+      return await db.transaction(async (tx) => {
+        const now = new Date()
+        const chatId = payload.chat.id
+        const members = payload.members ?? []
+        const characterId = pickCharacterId(members)
+
+        const existingChat = await tx.query.chats.findFirst({
+          where: eq(schema.chats.id, chatId),
+        })
+
+        if (existingChat) {
+          const member = await tx.query.chatMembers.findFirst({
+            where: and(
+              eq(schema.chatMembers.chatId, chatId),
+              eq(schema.chatMembers.memberType, 'user'),
+              eq(schema.chatMembers.userId, userId),
+            ),
+          })
+
+          if (!member)
+            throw createForbiddenError()
+        }
+
+        if (!existingChat) {
+          await tx.insert(schema.chats).values({
+            id: chatId,
+            type: payload.chat.type ?? 'group',
+            title: payload.chat.title,
+            createdAt: payload.chat.createdAt ? new Date(payload.chat.createdAt) : now,
+            updatedAt: payload.chat.updatedAt ? new Date(payload.chat.updatedAt) : now,
+          })
+        }
+        else {
+          const updates: Partial<schema.NewChat> = {
+            updatedAt: payload.chat.updatedAt ? new Date(payload.chat.updatedAt) : now,
+          }
+
+          if (payload.chat.type)
+            updates.type = payload.chat.type
+          if (payload.chat.title !== undefined)
+            updates.title = payload.chat.title
+
+          await tx.update(schema.chats)
+            .set(updates)
+            .where(eq(schema.chats.id, chatId))
+        }
+
+        const desiredMembers: SyncChatMemberPayload[] = [
+          { type: 'user', userId },
+          ...members.filter(member => member.type !== 'user'),
+        ]
+
+        for (const member of desiredMembers) {
+          if (member.type === 'user' && !member.userId)
+            continue
+          if (member.type === 'character' && !member.characterId)
+            continue
+
+          const existingMember = await tx.query.chatMembers.findFirst({
+            where: and(
+              eq(schema.chatMembers.chatId, chatId),
+              eq(schema.chatMembers.memberType, member.type),
+              member.type === 'user'
+                ? eq(schema.chatMembers.userId, member.userId!)
+                : eq(schema.chatMembers.characterId, member.characterId!),
+            ),
+          })
+
+          if (!existingMember) {
+            await tx.insert(schema.chatMembers).values({
+              chatId,
+              memberType: member.type,
+              userId: member.type === 'user' ? member.userId : null,
+              characterId: member.type === 'character' ? member.characterId : null,
+            })
+          }
+        }
+
+        for (const message of payload.messages) {
+          const existing = await tx.query.messages.findFirst({
+            where: eq(schema.messages.id, message.id),
+          })
+
+          const senderId = resolveSenderId(message.role, userId, characterId)
+          const createdAt = message.createdAt ? new Date(message.createdAt) : now
+
+          if (existing) {
+            if (existing.chatId !== chatId)
+              throw createConflictError('Message already belongs to another chat')
+
+            await tx.update(schema.messages)
+              .set({
+                senderId,
+                role: message.role,
+                content: message.content,
+                updatedAt: now,
+              })
+              .where(eq(schema.messages.id, message.id))
+            continue
+          }
+
+          await tx.insert(schema.messages).values({
+            id: message.id,
+            chatId,
+            senderId,
+            role: message.role,
+            content: message.content,
+            mediaIds: [],
+            stickerIds: [],
+            createdAt,
+            updatedAt: now,
+          })
+        }
+
+        return { chatId }
+      })
+    },
+  }
+}
+
+export type ChatService = ReturnType<typeof createChatService>

packages/stage-ui/src/stores/chat.ts

@@ -170,7 +170,7 @@ export const useChatOrchestratorStore = defineStore('chat-orchestrator', () => {
         return
 
       const sessionMessagesForSend = chatSession.getSessionMessages(sessionId)
-      sessionMessagesForSend.push({ role: 'user', content: finalContent })
+      sessionMessagesForSend.push({ role: 'user', content: finalContent, createdAt: sendingCreatedAt })
       chatSession.persistSessionMessages(sessionId)
 
       const categorizer = createStreamingCategorizer(activeProvider.value)

packages/stage-ui/src/stores/chat/session-store.ts

@@ -1,5 +1,3 @@
-import type { SystemMessage } from '@xsai/shared-chat'
-
 import type { ChatHistoryItem } from '../../types/chat'
 import type { ChatSessionMeta, ChatSessionRecord, ChatSessionsExport, ChatSessionsIndex } from '../../types/chat-session'
 
@@ -7,12 +5,14 @@ import { nanoid } from 'nanoid'
 import { defineStore, storeToRefs } from 'pinia'
 import { computed, ref, watch } from 'vue'
 
+import { client } from '../../composables/api'
+import { useLocalFirstRequest } from '../../composables/use-local-first'
 import { chatSessionsRepo } from '../../database/repos/chat-sessions.repo'
 import { useAuthStore } from '../auth'
 import { useAiriCardStore } from '../modules/airi-card'
 
 export const useChatSessionStore = defineStore('chat-session', () => {
-  const { userId } = storeToRefs(useAuthStore())
+  const { userId, isAuthenticated } = storeToRefs(useAuthStore())
   const { activeCardId, systemPrompt } = storeToRefs(useAiriCardStore())
 
   const activeSessionId = ref<string>('')
@@ -27,6 +27,7 @@ export const useChatSessionStore = defineStore('chat-session', () => {
   let initializePromise: Promise<void> | null = null
 
   let persistQueue = Promise.resolve()
+  let syncQueue = Promise.resolve()
   const loadedSessions = new Set<string>()
   const loadingSessions = new Map<string, Promise<void>>()
 
@@ -47,17 +48,115 @@ export const useChatSessionStore = defineStore('chat-session', () => {
     return persistQueue
   }
 
+  function enqueueSync(task: () => Promise<void>) {
+    syncQueue = syncQueue.then(task, task)
+    return syncQueue
+  }
+
   function snapshotMessages(messages: ChatHistoryItem[]) {
     return JSON.parse(JSON.stringify(messages)) as ChatHistoryItem[]
   }
 
+  function extractMessageContent(message: ChatHistoryItem) {
+    if (typeof message.content === 'string')
+      return message.content
+    if (Array.isArray(message.content)) {
+      return message.content.map((part) => {
+        if (typeof part === 'string')
+          return part
+        if (part && typeof part === 'object' && 'text' in part)
+          return String(part.text ?? '')
+        return ''
+      }).join('')
+    }
+    return ''
+  }
+
+  function buildMessageId(sessionId: string, message: ChatHistoryItem, index: number) {
+    const createdAt = message.createdAt ?? 0
+    return `${sessionId}:${createdAt}:${index}`
+  }
+
+  function buildSyncMessages(sessionId: string, messages: ChatHistoryItem[]) {
+    return messages.map((message, index) => ({
+      id: buildMessageId(sessionId, message, index),
+      role: message.role,
+      content: extractMessageContent(message),
+      createdAt: message.createdAt,
+    }))
+  }
+
+  async function syncSessionToRemote(sessionId: string) {
+    let cachedRecord: ChatSessionRecord | null | undefined
+    const request = useLocalFirstRequest({
+      local: async () => {
+        cachedRecord = await chatSessionsRepo.getSession(sessionId)
+        return cachedRecord
+      },
+      remote: async () => {
+        if (!cachedRecord)
+          cachedRecord = await chatSessionsRepo.getSession(sessionId)
+        if (!cachedRecord)
+          return cachedRecord
+
+        const members: Array<
+          | { type: 'user', userId: string }
+          | { type: 'character', characterId: string }
+        > = [
+          { type: 'user', userId: userId.value },
+        ]
+
+        if (cachedRecord.meta.characterId) {
+          members.push({
+            type: 'character',
+            characterId: cachedRecord.meta.characterId,
+          })
+        }
+
+        const res = await client.api.chats.sync.$post({
+          json: {
+            chat: {
+              id: cachedRecord.meta.sessionId,
+              type: 'group',
+              title: cachedRecord.meta.title,
+              createdAt: cachedRecord.meta.createdAt,
+              updatedAt: cachedRecord.meta.updatedAt,
+            },
+            members,
+            messages: buildSyncMessages(cachedRecord.meta.sessionId, cachedRecord.messages),
+          },
+        })
+
+        if (!res.ok)
+          throw new Error('Failed to sync chat session')
+        return cachedRecord
+      },
+      allowRemote: () => isAuthenticated.value,
+      lazy: true,
+    })
+
+    await request.execute()
+  }
+
+  function scheduleSync(sessionId: string) {
+    void enqueueSync(async () => {
+      try {
+        await syncSessionToRemote(sessionId)
+      }
+      catch (error) {
+        console.warn('Failed to sync chat session', error)
+      }
+    })
+  }
+
   function generateInitialMessageFromPrompt(prompt: string) {
     const content = codeBlockSystemPrompt + mathSyntaxSystemPrompt + prompt
 
     return {
       role: 'system',
       content,
-    } satisfies SystemMessage
+      createdAt: Date.now(),
+    } satisfies ChatHistoryItem
   }
 
   function generateInitialMessage() {
@@ -113,6 +212,7 @@ export const useChatSessionStore = defineStore('chat-session', () => {
 
     await enqueuePersist(() => chatSessionsRepo.saveSession(sessionId, record))
     await persistIndex()
+    scheduleSync(sessionId)
   }
 
   function persistSessionMessages(sessionId: string) {
@@ -178,8 +278,10 @@ export const useChatSessionStore = defineStore('chat-session', () => {
       characterIndex.activeSessionId = sessionId
     index.value.characters[characterId] = characterIndex
 
-    await enqueuePersist(() => chatSessionsRepo.saveSession(sessionId, { meta, messages: initialMessages }))
+    const record: ChatSessionRecord = { meta, messages: initialMessages }
+    await enqueuePersist(() => chatSessionsRepo.saveSession(sessionId, record))
     await persistIndex()
+    scheduleSync(sessionId)
 
     if (options?.setActive !== false)
       activeSessionId.value = sessionId