From 79c1c1d0c04cab137562e7b7f67d01e1d7352bc2 Mon Sep 17 00:00:00 2001 From: Xpert Agent Date: Mon, 16 Mar 2026 03:52:50 +0000 Subject: [PATCH] feat(authorization): add bulk import for assigning agent managers Add bulk import capability to the Agent Authorization page, allowing admins to assign an agent to multiple users at once by uploading a CSV file. Changes: - Frontend: Add 'Bulk Import' button and dialog component for CSV upload - Frontend: Preview found users before confirming the import - Backend: Add CSV upload endpoint with encoding detection for Chinese support - Backend: Add bulk add managers endpoint to handle multiple user assignments The CSV file must contain an 'email' column. Users are matched by their email addresses in the system. Fixes #417 --- .../src/app/@core/services/xpert.service.ts | 10 + .../authorization.component.html | 49 +- .../authorization/authorization.component.ts | 19 +- .../bulk-import/bulk-import.component.html | 183 ++ .../bulk-import/bulk-import.component.scss | 3 + .../bulk-import/bulk-import.component.ts | 92 + .../server-ai/src/xpert/xpert.controller.ts | 1856 +++++++++-------- packages/server-ai/src/xpert/xpert.service.ts | 913 ++++---- 8 files changed, 1833 insertions(+), 1292 deletions(-) create mode 100644 apps/cloud/src/app/features/xpert/xpert/authorization/bulk-import/bulk-import.component.html create mode 100644 apps/cloud/src/app/features/xpert/xpert/authorization/bulk-import/bulk-import.component.scss create mode 100644 apps/cloud/src/app/features/xpert/xpert/authorization/bulk-import/bulk-import.component.ts diff --git a/apps/cloud/src/app/@core/services/xpert.service.ts b/apps/cloud/src/app/@core/services/xpert.service.ts index a7904e7c38..ed18ac9ea2 100644 --- a/apps/cloud/src/app/@core/services/xpert.service.ts +++ b/apps/cloud/src/app/@core/services/xpert.service.ts @@ -162,6 +162,16 @@ export class XpertAPIService extends XpertWorkspaceBaseCrudService { return this.httpClient.delete(this.apiBaseUrl + `/${id}/managers/${userId}`) } + uploadAndParseManagersCsv(id: string, file: File) { + const formData = new FormData() + formData.append('file', file) + return this.httpClient.post(`${this.apiBaseUrl}/${id}/managers/bulk/upload`, formData) + } + + bulkAddManagers(id: string, userIds: string[]) { + return this.httpClient.put(this.apiBaseUrl + `/${id}/managers/bulk`, userIds) + } + getMyCopilots(relations?: string[]) { return this.getMyAll({ relations, diff --git a/apps/cloud/src/app/features/xpert/xpert/authorization/authorization.component.html b/apps/cloud/src/app/features/xpert/xpert/authorization/authorization.component.html index 508f267e26..517cb160d3 100644 --- a/apps/cloud/src/app/features/xpert/xpert/authorization/authorization.component.html +++ b/apps/cloud/src/app/features/xpert/xpert/authorization/authorization.component.html @@ -10,22 +10,41 @@
{{ xpert()?.title || xpert()?.name }}
-
- +
- {{ 'PAC.ACTIONS.Add' | translate: {Default: 'Add'} }} + + {{ 'PAC.Xpert.BulkImport' | translate: {Default: 'Bulk Import'} }} +
+
+ + {{ 'PAC.ACTIONS.Add' | translate: {Default: 'Add'} }} +
diff --git a/apps/cloud/src/app/features/xpert/xpert/authorization/authorization.component.ts b/apps/cloud/src/app/features/xpert/xpert/authorization/authorization.component.ts index e4cff3d944..6c073fd3ec 100644 --- a/apps/cloud/src/app/features/xpert/xpert/authorization/authorization.component.ts +++ b/apps/cloud/src/app/features/xpert/xpert/authorization/authorization.component.ts @@ -13,6 +13,7 @@ import { XpertComponent } from '../xpert.component' import { derivedAsync } from 'ngxtension/derived-async' import { UserProfileInlineComponent, UserRoleSelectComponent } from 'apps/cloud/src/app/@shared/user' import { Dialog } from '@angular/cdk/dialog' +import { XpertManagerBulkImportComponent } from './bulk-import/bulk-import.component' @Component({ standalone: true, @@ -29,7 +30,8 @@ import { Dialog } from '@angular/cdk/dialog' DragDropModule, MatTooltipModule, NgmSpinComponent, - UserProfileInlineComponent + UserProfileInlineComponent, + XpertManagerBulkImportComponent ] }) export class XpertAuthorizationComponent { @@ -73,6 +75,21 @@ export class XpertAuthorizationComponent { .subscribe() } + openBulkImport() { + this.#dialog + .open(XpertManagerBulkImportComponent, { + data: { + xpertId: this.xpertComponent.xpertId() + } + }) + .closed + .subscribe((result) => { + if (result) { + this.refresh$.next() + } + }) + } + addManagers(users: IUser[]) { this.loading.set(true) const newMembers = users.filter((u) => !this.members().some((_) => _.user.id === u.id)) diff --git a/apps/cloud/src/app/features/xpert/xpert/authorization/bulk-import/bulk-import.component.html b/apps/cloud/src/app/features/xpert/xpert/authorization/bulk-import/bulk-import.component.html new file mode 100644 index 0000000000..281dc29731 --- /dev/null +++ b/apps/cloud/src/app/features/xpert/xpert/authorization/bulk-import/bulk-import.component.html @@ -0,0 +1,183 @@ +
+ {{'PAC.Xpert.BulkImportManagers' | translate: {Default: 'Bulk Import Managers'} }} +
+
+ +
+
+
+ +
+
+
+ + @if (file()) { +
+ {{file().name}} +
+ +
+ +
+
+ } @else { +
+ {{'PAC.Xpert.DragDropCsv' | translate: {Default: 'Drag and drop your CSV file here, or '} }} + + {{'PAC.Xpert.Browse' | translate: {Default: 'Browse'} }} + +
+ } +
+
+
+
+ + +
+
+ {{'PAC.Xpert.ManagersCsvMustContain' | translate: {Default: 'The CSV file must contain an email column. Users will be found by their email address:'} }} +
+
+ + + + + + + + + + + + + + +
email
user1@example.com
user2@example.com
+
+ +
+ + {{'PAC.Xpert.DownloadTemplateHere' | translate: {Default: 'Download the template here'} }} +
+ +
+ + + @if (users().length > 0) { +
+
+ {{'PAC.Xpert.FoundUsers' | translate: {Default: 'Found users:'} }} {{ users().length }} +
+
+ + + + + + + + + @for (user of users(); track user.id) { + + + + + } + +
{{ 'PAC.KEY_WORDS.User' | translate: {Default: 'User'} }}{{ 'PAC.KEY_WORDS.Email' | translate: {Default: 'Email'} }}
{{ user.fullName || user.firstName + ' ' + user.lastName || user.username }}{{ user.email }}
+
+
+ } + + @if (loading()) { + + } +
+ + +
+ + +
diff --git a/apps/cloud/src/app/features/xpert/xpert/authorization/bulk-import/bulk-import.component.scss b/apps/cloud/src/app/features/xpert/xpert/authorization/bulk-import/bulk-import.component.scss new file mode 100644 index 0000000000..5d4e87f30f --- /dev/null +++ b/apps/cloud/src/app/features/xpert/xpert/authorization/bulk-import/bulk-import.component.scss @@ -0,0 +1,3 @@ +:host { + display: block; +} diff --git a/apps/cloud/src/app/features/xpert/xpert/authorization/bulk-import/bulk-import.component.ts b/apps/cloud/src/app/features/xpert/xpert/authorization/bulk-import/bulk-import.component.ts new file mode 100644 index 0000000000..87c6e46903 --- /dev/null +++ b/apps/cloud/src/app/features/xpert/xpert/authorization/bulk-import/bulk-import.component.ts @@ -0,0 +1,92 @@ +import { DIALOG_DATA, DialogRef } from '@angular/cdk/dialog' +import { CdkMenuModule } from '@angular/cdk/menu' +import { CommonModule } from '@angular/common' +import { ChangeDetectionStrategy, Component, inject, signal } from '@angular/core' +import { FormsModule } from '@angular/forms' +import { getErrorMessage, injectToastr, IUser, XpertAPIService } from '@cloud/app/@core' +import { NgmDndDirective, OverlayAnimation1 } from '@metad/core' +import { NgmSpinComponent } from '@metad/ocap-angular/common' +import { TranslateModule } from '@ngx-translate/core' +import { firstValueFrom } from 'rxjs' + +@Component({ + standalone: true, + imports: [CommonModule, FormsModule, TranslateModule, CdkMenuModule, NgmDndDirective, NgmSpinComponent], + selector: 'xpert-manager-bulk-import', + templateUrl: './bulk-import.component.html', + styleUrl: './bulk-import.component.scss', + animations: [OverlayAnimation1], + changeDetection: ChangeDetectionStrategy.OnPush +}) +export class XpertManagerBulkImportComponent { + readonly #data = inject<{ xpertId: string }>(DIALOG_DATA) + readonly #dialogRef = inject(DialogRef) + readonly #xpertAPI = inject(XpertAPIService) + readonly #toastr = injectToastr() + + readonly xpertId = signal(this.#data.xpertId) + readonly file = signal(null) + readonly users = signal([]) + readonly loading = signal(false) + + close() { + this.#dialogRef.close() + } + + downloadTemplate() { + const csvContent = 'email\nuser1@example.com\nuser2@example.com\n' + const bom = '\uFEFF' + const blob = new Blob([bom + csvContent], { type: 'text/csv;charset=utf-8;' }) + const url = URL.createObjectURL(blob) + const a = document.createElement('a') + a.href = url + a.download = 'managers-template.csv' + a.click() + URL.revokeObjectURL(url) + } + + async onFileDropped(event: FileList) { + if (event.length > 0) { + await this.onFile(event.item(0)) + } + } + + async onFile(file: File) { + try { + this.loading.set(true) + this.file.set(file) + // Call backend API to parse CSV with proper encoding detection + const parsedUsers = await firstValueFrom( + this.#xpertAPI.uploadAndParseManagersCsv(this.xpertId(), file) + ) + this.users.set(parsedUsers) + } catch (err) { + this.#toastr.error(getErrorMessage(err)) + } finally { + this.loading.set(false) + } + } + + async onFileSelected(event: Event) { + const input = event.target as HTMLInputElement + if (input.files && input.files.length > 0) { + const file = input.files[0] + this.onFile(file) + } + } + + async upload() { + this.loading.set(true) + this.#xpertAPI.bulkAddManagers(this.xpertId(), this.users().map(u => u.id)).subscribe({ + next: (response) => { + this.loading.set(false) + this.#toastr.success('PAC.Messages.SavedSuccessfully', { Default: 'Saved Successfully' }) + this.#dialogRef.close(true) + }, + error: (error) => { + this.loading.set(false) + this.#toastr.error(getErrorMessage(error)) + } + }) + } +} diff --git a/packages/server-ai/src/xpert/xpert.controller.ts b/packages/server-ai/src/xpert/xpert.controller.ts index 56685c5b43..558b5782c9 100644 --- a/packages/server-ai/src/xpert/xpert.controller.ts +++ b/packages/server-ai/src/xpert/xpert.controller.ts @@ -1,42 +1,42 @@ import { IChatConversation, IIntegration, IXpert, LanguagesEnum, LongTermMemoryTypeEnum, RolesEnum, TChatApi, TChatApp, TChatOptions, TChatRequest, TMemoryQA, TMemoryUserProfile, TXpertTeamDraft, UserType, xpertLabel } from '@metad/contracts' import { - CrudController, - OptionParams, - PaginationParams, - ParseJsonPipe, - RequestContext, - RoleGuard, - Roles, - TransformInterceptor, - UserPublicDTO, - UseValidationPipe, - UUIDValidationPipe, - UserCreateCommand, - Public, - TimeZone, + CrudController, + OptionParams, + PaginationParams, + ParseJsonPipe, + RequestContext, + RoleGuard, + Roles, + TransformInterceptor, + UserPublicDTO, + UseValidationPipe, + UUIDValidationPipe, + UserCreateCommand, + Public, + TimeZone, } from '@metad/server-core' import { - Body, - Controller, - Delete, - Get, - Header, - HttpCode, - HttpStatus, - Logger, - Param, - Post, - Put, - Query, - Sse, - UseInterceptors, - UseGuards, - HttpException, - ForbiddenException, - InternalServerErrorException, - Res, - NotFoundException, - BadRequestException + Body, + Controller, + Delete, + Get, + Header, + HttpCode, + HttpStatus, + Logger, + Param, + Post, + Put, + Query, + Sse, + UseInterceptors, + UseGuards, + HttpException, + ForbiddenException, + InternalServerErrorException, + Res, + NotFoundException, + BadRequestException } from '@nestjs/common' import { FileInterceptor } from '@nestjs/platform-express' import { UploadedFile } from '@metad/contracts' @@ -80,806 +80,988 @@ import { AGENT_CHAT_MESSAGE_TYPE } from '../handoff/local-sync-task.service' @UseInterceptors(TransformInterceptor) @Controller() export class XpertController extends CrudController { - readonly #logger = new Logger(XpertController.name) - constructor( - private readonly service: XpertService, - private readonly storeService: CopilotStoreService, - private readonly environmentService: EnvironmentService, - private readonly i18n: I18nService, - private readonly commandBus: CommandBus, - private readonly queryBus: QueryBus - ) { - super(service) - } - - @UseGuards(RoleGuard) - @Roles(RolesEnum.ADMIN, RolesEnum.SUPER_ADMIN) - @Get() - async getAll(@Query('data', ParseJsonPipe) params: Partial>, @Query('published') published?: boolean) { - const { where, ...rest } = params - if (published) { - where.version = Not(IsNull()) - } - - const result = await this.service.findAll({...rest, where}) - return { - ...result, - items: result.items.map((item) => new XpertPublicDTO(item)) - } - } - - @UseGuards(WorkspaceGuard) - @Get('by-workspace/:workspaceId') - async getAllByWorkspace( - @Param('workspaceId') workspaceId: string, - @Query('data', ParseJsonPipe) data: PaginationParams, - @Query('published') published?: boolean - ) { - const result = await this.service.getAllByWorkspace(workspaceId, data, published, RequestContext.currentUser()) - return { - ...result, - items: result.items.map((item) => new XpertPublicDTO(item)) - } - } - - @Get('my') - async getMyAll(@Query('data', ParseJsonPipe) params: PaginationParams,) { - return this.service.getMyAll(params) - } - - @Get('validate') - async validateName(@Query('name') name: string) { - return this.service.validateName(name) - } - - @UseValidationPipe({ transform: true }) - @Post('import') - async importDSL(@Body() dsl: XpertDraftDslDTO) { - try { - return await this.commandBus.execute(new XpertImportCommand(dsl)) - } catch (error) { - throw new HttpException( - error.message, - HttpStatus.INTERNAL_SERVER_ERROR - ) - } - } - - @UseGuards(RoleGuard) - @Roles(RolesEnum.ADMIN, RolesEnum.SUPER_ADMIN) - @Get('select-options') - async getSelectOptions() { - const { items } = await this.getAll({where: {latest: true, }}, true, ) - return items.map((item) => ({ - value: item.id, - label: xpertLabel(item) - })) - } - - @Get('triggers/providers') - async getTriggerProviders() { - return this.service.getTriggerProviders() - } - - @Get('sandbox/providers') - async getSandboxProviders() { - return this.service.getSandboxProviders() - } - - @Get('slug/:slug') - async getOneBySlug(@Param('slug') slug: string,) { - const xpert = await this.service.findBySlug(slug, ['agent', 'agents']) - return xpert ? new XpertPublicDTO(xpert) : null - } - - @UseGuards(XpertGuard) - @Get(':id/export') - async exportDSL( - @Param('id') xpertId: string, - @Query('isDraft') isDraft: string, - @Query('includeMemory') includeMemory: string, - @Query('data', ParseJsonPipe) params: PaginationParams) { - try { - const xpert = await this.commandBus.execute(new XpertExportCommand(xpertId, isDraft === 'true', includeMemory === 'true')) - return { - data: yaml.stringify(instanceToPlain(xpert)) - } - } catch(err) { - throw new InternalServerErrorException(err.message) - } - } - - @UseGuards(XpertGuard) - @Get(':id/team') - async getTeam(@Param('id') id: string, @Query('data', ParseJsonPipe) data: OptionParams) { - return this.service.getTeam(id, data) - } - - @Get(':id/version') - async allVersions(@Param('id') id: string) { - return this.service.allVersions(id) - } - - @Post(':id/latest') - async setAsLatest(@Param('id') id: string) { - return this.service.setAsLatest(id) - } - - @UseGuards(XpertGuard) - @Post(':id/draft') - async saveDraft(@Param('id') id: string, @Body() draft: TXpertTeamDraft) { - // todo Check if you have permission to edit this xpert role - draft.savedAt = new Date() - // Save draft - return await this.service.saveDraft(id, draft) - } - - @UseGuards(XpertGuard) - @Put(':id/draft') - async updateDraft(@Param('id') id: string, @Body() draft: TXpertTeamDraft) { - // todo Check if you have permission to edit this xpert role - draft.savedAt = new Date() - // Save draft - return await this.service.updateDraft(id, draft) - } - - @UseGuards(XpertGuard) - @Post(':id/publish') - async publish( - @Param('id') id: string, - @Query('newVersion') newVersion: string, - @Body() body: {environmentId: string; releaseNotes: string} - ) { - return this.service.publish(id, newVersion === 'true', body.environmentId, body.releaseNotes) - } - - /** - * @deprecated use workflow trigger instead - */ - @UseGuards(XpertGuard) - @Post(':id/publish/integration') - async publishIntegration(@Param('id') id: string, @Body() integration: Partial) { - return this.commandBus.execute(new XpertPublishIntegrationCommand(id, integration)) - } - - /** - * @deprecated use workflow trigger instead - */ - @UseGuards(XpertGuard) - @Delete(':id/publish/integration/:integration') - async deleteIntegration(@Param('id') id: string, @Param('integration') integration: string,) { - return this.commandBus.execute(new XpertDelIntegrationCommand(id, integration)) - } - - @Get(':id/diagram') - async getDiagram( - @Res() res: Response, - @Param('id') id: string, - @Query('isDraft') isDraft: string, - @Query('agentKey') agentKey: string,) { - try { - const imageData = await this.commandBus.execute(new XpertExportDiagramCommand(id, isDraft === 'true', agentKey)) - res.setHeader('Content-Type', 'image/jpeg') - res.send(Buffer.from(await imageData.arrayBuffer())) - } catch (err) { - console.error(err) - throw new InternalServerErrorException(err.message) - } - } - - @Get(':id/executions') - async getExecutions( - @Param('id') id: string, - @Query('$order', ParseJsonPipe) order?: PaginationParams['order'] - ) { - return this.queryBus.execute(new FindExecutionsByXpertQuery(id, { order })) - } - - @Header('content-type', 'text/event-stream') - @Header('Connection', 'keep-alive') - @Header('Cache-Control', 'no-cache') - @Post(':id/chat') - @Sse() - async chat(@Res() res: Response, - @Param('id') id: string, - @I18nLang() language: LanguagesEnum, - @TimeZone() timeZone: string, - @Body() - body: { - request: TChatRequest - options: TChatOptions - } - ) { - let environment = null - if (body.request.environmentId) { - environment = await this.environmentService.findOne(body.request.environmentId) - } - const observable = await this.enqueueXpertChatTask( - body.request, - { - ...body.options, - xpertId: id, - environment, - language, - timeZone, - from: 'debugger' - } - ) - return observable.pipe( - // Add an operator to send a comment event periodically (30s) to keep the connection alive - keepAlive(30000), - takeUntilClose(res) - ) - } - - @ApiOperation({ summary: 'Delete record' }) - @ApiResponse({ - status: HttpStatus.NO_CONTENT, - description: 'The record has been successfully deleted' - }) - @ApiResponse({ - status: HttpStatus.NOT_FOUND, - description: 'Record not found' - }) - @HttpCode(HttpStatus.ACCEPTED) - @UseGuards(XpertGuard) - @Delete(':id') - async delete( - @Param('id', UUIDValidationPipe) id: string, - ): Promise { - return this.commandBus.execute(new XpertDeleteCommand(id)) - } - - @Get(':id/managers') - async getManagers(@Param('id') id: string) { - const xpert = await this.service.findOne(id, { relations: ['managers'] }) - return xpert.managers.map((u) => new UserPublicDTO(u)) - } - - @Put(':id/managers') - async updateManagers(@Param('id') id: string, @Body() ids: string[]) { - return this.service.updateManagers(id, ids) - } - - @Delete(':id/managers/:userId') - async removeManager(@Param('id') id: string, @Param('userId') userId: string) { - await this.service.removeManager(id, userId) - } - - @Get(':id/memory') - async getAllMemory(@Param('id') id: string, @Query('types') types: string) { - const _types = types?.split(':').filter((_) => !!_) - return this.service.findAllMemory(id, _types) - } - - @Post(':id/memory/bulk') - async createBulkMemory(@Param('id') id: string, @Body() body: { - type: LongTermMemoryTypeEnum; memories: Array - }) { - return this.service.createBulkMemories(id, body) - } - - /** - * @todo Refactoring is required - */ - @Post(':id/memory/bulk/upload') - @UseInterceptors( - FileInterceptor('file', { - storage: new FileStorage().storage({ - dest: path.join('temp'), // Store in a temporary directory - prefix: 'memory-csv-upload' - }) - }) - ) - @ApiOperation({ summary: 'Upload and parse CSV file for bulk memory import' }) - @ApiResponse({ - status: HttpStatus.OK, - description: 'Memories parsed from CSV file' - }) - async uploadAndParseCsv( - @Param('id') id: string, - @Body() body: { type: LongTermMemoryTypeEnum }, - @UploadedFileStorage() file: UploadedFile - ): Promise> { - const type = body.type - if (!file) { - throw new BadRequestException('No file uploaded') - } - - const { path: filePath, mimetype, originalname } = file - - if (mimetype !== 'text/csv' && !originalname.endsWith('.csv')) { - throw new BadRequestException('Only CSV files are supported') - } - - try { - // Read file buffer - const buffer = await fsPromises.readFile(filePath) - - // Function to check if string contains valid Chinese characters - const containsValidChinese = (str: string): boolean => { - return /[\u4e00-\u9fa5]/.test(str) - } - - // Function to detect if string looks like mis-decoded UTF-8 - const looksLikeMisDecoded = (str: string): boolean => { - return /[´°µÄÂÈøò·¢]{3,}/.test(str) - } - - // Function to check if decoding is likely correct - const isValidDecoding = (decoded: string, encoding: string): boolean => { - if (decoded.includes('\uFFFD')) { - return false - } - - if (encoding === 'utf8' || encoding === 'utf-8') { - try { - Buffer.from(decoded, 'utf8') - return true - } catch { - return false - } - } - - return true - } - - // Try different encodings - const encodingsToTry = ['utf8', 'gbk', 'gb18030', 'gb2312', 'big5'] - - // Check for BOM - let startOffset = 0 - if (buffer[0] === 0xEF && buffer[1] === 0xBB && buffer[2] === 0xBF) { - startOffset = 3 - } - - const bufferToDecode = startOffset > 0 ? buffer.slice(startOffset) : buffer - - let content: string | null = null - let lastError: Error | null = null - - for (const encoding of encodingsToTry) { - try { - const decoded = iconv.decode(bufferToDecode, encoding) - - if (!isValidDecoding(decoded, encoding)) { - continue - } - - if (looksLikeMisDecoded(decoded)) { - continue - } - - if (containsValidChinese(decoded)) { - content = decoded - break - } - - if (encoding === 'utf8' && !content && decoded.length > 0) { - content = decoded - } else if (!content && decoded.length > 0 && !decoded.match(/[^\x00-\x7F]/)) { - content = decoded - } - } catch (err) { - lastError = err - } - } - - if (!content) { - try { - content = iconv.decode(bufferToDecode, 'utf8') - } catch (err) { - throw new BadRequestException(`Failed to decode CSV file. Tried encodings: ${encodingsToTry.join(', ')}. Error: ${lastError?.message || err.message}`) - } - } - - if (!content) { - throw new BadRequestException('Failed to decode CSV file with any known encoding') - } - - // Parse CSV with XLSX - const workbook = XLSX.read(content, { - type: 'string', - codepage: 65001 // UTF-8 codepage - }) - - const sheet = workbook.Sheets[workbook.SheetNames[0]] - const jsonData = XLSX.utils.sheet_to_json(sheet) as any[] - - // Map to memory format based on type - const memories: Array = jsonData.map((row: any) => { - if (type === LongTermMemoryTypeEnum.QA) { - return { - question: row.question || row.问题 || row.問題, - answer: row.answer || row.答案 - } as TMemoryQA - } else if (type === LongTermMemoryTypeEnum.PROFILE) { - return { - profile: row.profile || row.档案 || row.檔案, - context: row.context || row.上下文 - } as TMemoryUserProfile - } - return row - }) - - // Clean up temporary file - try { - await fsPromises.unlink(filePath) - } catch (err) { - // Ignore cleanup errors - } - - return memories - } catch (error) { - // Clean up temporary file on error - try { - await fsPromises.unlink(filePath) - } catch (err) { - // Ignore cleanup errors - } - throw new BadRequestException(`Failed to parse CSV file: ${error.message}`) - } - } - - @Post(':id/memory') - async createMemory(@Param('id') id: string, @Body() body: {type: LongTermMemoryTypeEnum; value: TMemoryQA | TMemoryUserProfile}) { - return this.service.createMemory(id, body) - } - - @Post(':id/memory/search') - async searchMemory(@Param('id') id: string, @Body() body: {type: LongTermMemoryTypeEnum; text: string; isDraft?: boolean;}) { - try { - return await this.queryBus.execute(new SearchXpertMemoryQuery(id, body)) - } catch(err) { - throw new HttpException(getErrorMessage(err), HttpStatus.INTERNAL_SERVER_ERROR) - } - } - - @Delete(':id/memory') - async clearMemory(@Param('id') id: string,) { - try { - return await this.storeService.delete({prefix: Like(`${id}%`)}) - } catch(err) { - throw new HttpException(getErrorMessage(err), HttpStatus.INTERNAL_SERVER_ERROR) - } - } - - @Get(':id/variables') - async getVariables( - @Param('id') id: string, - @Query('environment') environmentId: string, - @Query('inputs') inputs: string - ) { - try { - const inputKeys = inputs ? inputs.split(',').map((value) => value.trim()).filter(Boolean) : undefined - return await this.queryBus.execute( - new XpertAgentVariablesQuery({xpertId: id, isDraft: true, environmentId, inputs: inputKeys}) - ) - } catch (err) { - throw new HttpException(getErrorMessage(err), HttpStatus.INTERNAL_SERVER_ERROR) - } - } - - @Get(':id/agent/:agent/variables') - async getAgentVariables( - @Param('id') id: string, - @Param('agent') agentKey: string, - @Query('environment') environmentId: string, - @Query('isDraft') isDraft: string, - @Query('type') type: 'input' | 'output', - @Query('inputs') inputs: string - ) { - try { - const inputKeys = inputs ? inputs.split(',').map((value) => value.trim()).filter(Boolean) : undefined - return await this.queryBus.execute(new XpertAgentVariablesQuery({ - xpertId: id, type, nodeKey: agentKey, isDraft: !(isDraft === 'false'), environmentId, inputs: inputKeys})) - } catch (err) { - throw new HttpException(getErrorMessage(err), HttpStatus.INTERNAL_SERVER_ERROR) - } - } - - /** - * Get available variables for a workflow node - */ - @Get(':id/workflow/:key/variables') - async getWorkflowVariables( - @Param('id') id: string, - @Param('key') nodeKey: string, - @Query('environment') environmentId: string, - @Query('type') type: 'input' | 'output', - @Query('inputs') inputs: string - ) { - try { - const inputKeys = inputs ? inputs.split(',').map((value) => value.trim()).filter(Boolean) : undefined - return await this.queryBus.execute( - new XpertAgentVariablesQuery({xpertId: id, type, nodeKey, isDraft: true, environmentId, inputs: inputKeys}) - ) - } catch (err) { - throw new HttpException(getErrorMessage(err), HttpStatus.INTERNAL_SERVER_ERROR) - } - } - - @Put(':id/api') - async updateChatApi(@Param('id') id: string, @Body() api: Partial) { - const xpert = await this.service.findOne(id) - await this.service.update(id, {api: {...(xpert.api ?? {}), ...api}}) - if (!api.disabled && !xpert.userId) { - const user = await this.commandBus.execute(new UserCreateCommand({ - username: xpert.slug, - type: UserType.COMMUNICATION, - preferredLanguage: LanguagesEnum.English, - hash: uuidv4(), - })) - await this.service.update(id, {user}) - } - } - - @Put(':id/app') - async updateChatApp(@Param('id') id: string, @Body() app: Partial) { - const xpert = await this.service.findOne(id) - await this.service.update(id, {app: {...(xpert.app ?? {}), ...app}}) - if (app.enabled && !xpert.userId) { - const user = await this.commandBus.execute(new UserCreateCommand({ - username: xpert.slug, - type: UserType.COMMUNICATION, - preferredLanguage: LanguagesEnum.English, - hash: uuidv4(), - })) - await this.service.update(id, {user}) - } - } - - @Post(':id/duplicate') - async duplicate(@Param('id') id: string, @Body() body: {basic: Partial; isDraft: boolean}) { - try { - const xpertDto = await this.commandBus.execute(new XpertExportCommand(id, body.isDraft, false)) - const dsl = instanceToPlain(xpertDto) - return await this.commandBus.execute(new XpertImportCommand({...dsl, team: {...dsl.team, ...body.basic}})) - } catch(err) { - throw new InternalServerErrorException(err.message) - } - } - - // Conversations - - @UseGuards(XpertGuard) - @Get(':id/conversations') - async getConversations( - @Param('id') id: string, - @Query('data', ParseJsonPipe) data: PaginationParams, - @Query('start') start: string, - @Query('end') end: string - ) { - const {where} = data - const result = await this.queryBus.execute(new ChatConversationLogsQuery( - { - ...data, - where: { - ...(where ?? {}), - xpertId: id, - createdAt: start ? Between(new Date(start), new Date(end)) : LessThanOrEqual(new Date(end)) - }, - }, - )) - return { - ...result, - items: result.items.map((item) => new ChatConversationPublicDTO(item)) - } - } - - // Public App - - @Public() - @UseGuards(AnonymousXpertAuthGuard) - @Get(':name/app') - async getChatApp(@Param('name') name: string,) { - const xpert = await this.service.findBySlug(name) - if (!xpert) { - throw new NotFoundException(`Not found xpert '${name}'`) - } - - if (!xpert.app?.enabled) { - throw new ForbiddenException(name) - } - - return new XpertPublicDTO(xpert) - } - - private getPublicUserCondition() { - const userId = RequestContext.currentUserId() - const fromEndUserId = ((RequestContext.currentRequest())).cookies['anonymous.id'] - return userId ? fromEndUserId ? {createdById: userId, fromEndUserId} : {createdById: userId} : {fromEndUserId} - } - - @Public() - @UseGuards(AnonymousXpertAuthGuard) - @Get(':name/conversation/:id') - async getAppConversation(@Param('name') name: string, @Param('id') id: string, - @Query('$relations', ParseJsonPipe) relations?: PaginationParams['relations'], - @Query('$select', ParseJsonPipe) select?: PaginationParams['select'], - ) { - const conversation = await this.queryBus.execute(new GetChatConversationQuery({ - id, - ...this.getPublicUserCondition(), - }, relations)) - return conversation - } - - @Public() - @UseGuards(AnonymousXpertAuthGuard) - @Delete(':name/conversation/:id') - async deleteAppConversation(@Param('name') slug: string, @Param('id') id: string,) { - await this.queryBus.execute(new GetChatConversationQuery({id, ...this.getPublicUserCondition(),})) - await this.commandBus.execute(new ChatConversationDeleteCommand({id, ...this.getPublicUserCondition(),})) - } - - @Public() - @UseGuards(AnonymousXpertAuthGuard) - @Get(':name/conversation') - async getAppConversations(@Param('name') slug: string, - @Query('data', ParseJsonPipe) paginationOptions?: PaginationParams, - ) { - const xpert = await this.service.findBySlug(slug) - const conversation = await this.queryBus.execute(new FindChatConversationQuery({ - ...(paginationOptions.where ?? {}), - ...this.getPublicUserCondition(), - xpertId: xpert.id - }, paginationOptions)) - return conversation - } - - @Public() - @UseGuards(AnonymousXpertAuthGuard) - @Put(':name/conversation/:id') - async updateAppConversation(@Param('name') slug: string, @Param('id') id: string, @Body() entity: Partial) { - await this.queryBus.execute(new FindChatConversationQuery({id, ...this.getPublicUserCondition(),})) - await this.commandBus.execute(new ChatConversationUpsertCommand({ - id, - ...entity - })) - } - - @Public() - @UseGuards(AnonymousXpertAuthGuard) - @Get(':name/conversation/:id/feedbacks') - async getAppFeedbacks(@Param('name') name: string, @Param('id') id: string, - @Query('$relations', ParseJsonPipe) relations?: PaginationParams['relations'], - @Query('$select', ParseJsonPipe) select?: PaginationParams['select'], - ) { - const conversation = await this.queryBus.execute(new FindChatConversationQuery({id, ...this.getPublicUserCondition(),}, {relations})) - return await this.queryBus.execute(new FindMessageFeedbackQuery({conversationId: conversation.id}, relations)) - } - - @Public() - @UseGuards(AnonymousXpertAuthGuard) - @Header('content-type', 'text/event-stream') - @Header('Connection', 'keep-alive') - @Post(':name/chat-app') - @Sse() - async chatApp(@Res() res: Response, @Param('name') name: string, @I18nLang() language: LanguagesEnum, - @Body() body: { request: TChatRequest; options: TChatOptions }) { - const fromEndUserId = ((RequestContext.currentRequest())).cookies['anonymous.id'] - const observable = await this.enqueueXpertChatTask( - body.request, - { - ...body.options, - language, - from: 'webapp', - fromEndUserId - } - ) - return observable.pipe( - // Add an operator to send a comment event periodically (30s) to keep the connection alive - keepAlive(30000), - takeUntilClose(res) - ) - } - - private async enqueueXpertChatTask( - request: TChatRequest, - options: NonNullable[1]> - ) { - const queueTaskId = `xpert-chat-${uuidv4()}` - const sessionKey = request.conversationId ?? options.messageId ?? queueTaskId - - return this.commandBus.execute( - new EnqueueAgentChatMessageCommand( - { - id: queueTaskId, - messageType: AGENT_CHAT_MESSAGE_TYPE, - tenantId: RequestContext.currentTenantId(), - organizationId: RequestContext.getOrganizationId(), - userId: RequestContext.currentUserId(), - sessionKey, - conversationId: request.conversationId, - executionId: options.execution?.id, - source: 'chat', - queueName: XPERT_HANDOFF_QUEUE, - businessKey: sessionKey, - traceId: options.messageId ?? queueTaskId - }, - async () => this.commandBus.execute(new XpertChatCommand(request, options)) - ) - ) - } - - // Statistics - - @UseGuards(RoleGuard) - @Roles(RolesEnum.ADMIN, RolesEnum.SUPER_ADMIN, RolesEnum.TRIAL) - @Get('statistics/xperts') - async getStatisticsXperts(@Query('start') start: string, @Query('end') end: string) { - return await this.queryBus.execute(new StatisticsXpertsQuery(start, end)) - } - - @UseGuards(RoleGuard) - @Roles(RolesEnum.ADMIN, RolesEnum.SUPER_ADMIN, RolesEnum.TRIAL) - @Get('statistics/xpert-conversations') - async getStatisticsXpertConversations(@Query('start') start: string, @Query('end') end: string) { - return await this.queryBus.execute(new StatisticsXpertConversationsQuery(start, end)) - } - - @UseGuards(RoleGuard) - @Roles(RolesEnum.ADMIN, RolesEnum.SUPER_ADMIN, RolesEnum.TRIAL) - @Get('statistics/xpert-messages') - async getStatisticsXpertMessages(@Query('start') start: string, @Query('end') end: string) { - return await this.queryBus.execute(new StatisticsXpertMessagesQuery(start, end)) - } - - @UseGuards(RoleGuard) - @Roles(RolesEnum.ADMIN, RolesEnum.SUPER_ADMIN, RolesEnum.TRIAL) - @Get('statistics/xpert-tokens') - async getStatisticsXpertTokens(@Query('start') start: string, @Query('end') end: string) { - return await this.queryBus.execute(new StatisticsXpertTokensQuery(start, end)) - } - - @UseGuards(RoleGuard) - @Roles(RolesEnum.ADMIN, RolesEnum.SUPER_ADMIN, RolesEnum.TRIAL) - @Get('statistics/xpert-integrations') - async getStatisticsXpertIntegrations(@Query('start') start: string, @Query('end') end: string) { - return await this.queryBus.execute(new StatisticsXpertIntegrationsQuery(start, end)) - } - - @UseGuards(XpertGuard) - @Get(':id/statistics/daily-conversations') - async getDailyConversations(@Param('id') id: string, @Query('start') start: string, @Query('end') end: string) { - return await this.queryBus.execute(new StatisticsDailyConvQuery(start, end, id)) - } - - @UseGuards(XpertGuard) - @Get(':id/statistics/daily-end-users') - async getDailyEndUsers(@Param('id') id: string, @Query('start') start: string, @Query('end') end: string) { - return await this.queryBus.execute(new StatisticsDailyEndUsersQuery(start, end, id)) - } - - @UseGuards(XpertGuard) - @Get(':id/statistics/average-session-interactions') - async getAverageSessionInteractions(@Param('id') id: string, @Query('start') start: string, @Query('end') end: string) { - return await this.queryBus.execute(new StatisticsAverageSessionInteractionsQuery(start, end, id)) - } - - @UseGuards(XpertGuard) - @Get(':id/statistics/daily-messages') - async getDailyMessages(@Param('id') id: string, @Query('start') start: string, @Query('end') end: string) { - return await this.queryBus.execute(new StatisticsDailyMessagesQuery(start, end, id)) - } - - @UseGuards(XpertGuard) - @Get(':id/statistics/tokens-per-second') - async getTokensPerSecond(@Param('id') id: string, @Query('start') start: string, @Query('end') end: string) { - return await this.queryBus.execute(new StatisticsTokensPerSecondQuery(start, end, id)) - } - - @UseGuards(XpertGuard) - @Get(':id/statistics/token-costs') - async getTokenCost(@Param('id') id: string, @Query('start') start: string, @Query('end') end: string) { - return await this.queryBus.execute(new StatisticsTokenCostQuery(start, end, id)) - } - - @UseGuards(XpertGuard) - @Get(':id/statistics/user-satisfaction-rate') - async getUserSatisfactionRate(@Param('id') id: string, @Query('start') start: string, @Query('end') end: string) { - return await this.queryBus.execute(new StatisticsUserSatisfactionRateQuery(start, end, id)) - } - + readonly #logger = new Logger(XpertController.name) + constructor( + private readonly service: XpertService, + private readonly storeService: CopilotStoreService, + private readonly environmentService: EnvironmentService, + private readonly i18n: I18nService, + private readonly commandBus: CommandBus, + private readonly queryBus: QueryBus + ) { + super(service) + } + + @UseGuards(RoleGuard) + @Roles(RolesEnum.ADMIN, RolesEnum.SUPER_ADMIN) + @Get() + async getAll(@Query('data', ParseJsonPipe) params: Partial>, @Query('published') published?: boolean) { + const { where, ...rest } = params + if (published) { + where.version = Not(IsNull()) + } + + const result = await this.service.findAll({...rest, where}) + return { + ...result, + items: result.items.map((item) => new XpertPublicDTO(item)) + } + } + + @UseGuards(WorkspaceGuard) + @Get('by-workspace/:workspaceId') + async getAllByWorkspace( + @Param('workspaceId') workspaceId: string, + @Query('data', ParseJsonPipe) data: PaginationParams, + @Query('published') published?: boolean + ) { + const result = await this.service.getAllByWorkspace(workspaceId, data, published, RequestContext.currentUser()) + return { + ...result, + items: result.items.map((item) => new XpertPublicDTO(item)) + } + } + + @Get('my') + async getMyAll(@Query('data', ParseJsonPipe) params: PaginationParams,) { + return this.service.getMyAll(params) + } + + @Get('validate') + async validateName(@Query('name') name: string) { + return this.service.validateName(name) + } + + @UseValidationPipe({ transform: true }) + @Post('import') + async importDSL(@Body() dsl: XpertDraftDslDTO) { + try { + return await this.commandBus.execute(new XpertImportCommand(dsl)) + } catch (error) { + throw new HttpException( + error.message, + HttpStatus.INTERNAL_SERVER_ERROR + ) + } + } + + @UseGuards(RoleGuard) + @Roles(RolesEnum.ADMIN, RolesEnum.SUPER_ADMIN) + @Get('select-options') + async getSelectOptions() { + const { items } = await this.getAll({where: {latest: true, }}, true, ) + return items.map((item) => ({ + value: item.id, + label: xpertLabel(item) + })) + } + + @Get('triggers/providers') + async getTriggerProviders() { + return this.service.getTriggerProviders() + } + + @Get('sandbox/providers') + async getSandboxProviders() { + return this.service.getSandboxProviders() + } + + @Get('slug/:slug') + async getOneBySlug(@Param('slug') slug: string,) { + const xpert = await this.service.findBySlug(slug, ['agent', 'agents']) + return xpert ? new XpertPublicDTO(xpert) : null + } + + @UseGuards(XpertGuard) + @Get(':id/export') + async exportDSL( + @Param('id') xpertId: string, + @Query('isDraft') isDraft: string, + @Query('includeMemory') includeMemory: string, + @Query('data', ParseJsonPipe) params: PaginationParams) { + try { + const xpert = await this.commandBus.execute(new XpertExportCommand(xpertId, isDraft === 'true', includeMemory === 'true')) + return { + data: yaml.stringify(instanceToPlain(xpert)) + } + } catch(err) { + throw new InternalServerErrorException(err.message) + } + } + + @UseGuards(XpertGuard) + @Get(':id/team') + async getTeam(@Param('id') id: string, @Query('data', ParseJsonPipe) data: OptionParams) { + return this.service.getTeam(id, data) + } + + @Get(':id/version') + async allVersions(@Param('id') id: string) { + return this.service.allVersions(id) + } + + @Post(':id/latest') + async setAsLatest(@Param('id') id: string) { + return this.service.setAsLatest(id) + } + + @UseGuards(XpertGuard) + @Post(':id/draft') + async saveDraft(@Param('id') id: string, @Body() draft: TXpertTeamDraft) { + // todo Check if you have permission to edit this xpert role + draft.savedAt = new Date() + // Save draft + return await this.service.saveDraft(id, draft) + } + + @UseGuards(XpertGuard) + @Put(':id/draft') + async updateDraft(@Param('id') id: string, @Body() draft: TXpertTeamDraft) { + // todo Check if you have permission to edit this xpert role + draft.savedAt = new Date() + // Save draft + return await this.service.updateDraft(id, draft) + } + + @UseGuards(XpertGuard) + @Post(':id/publish') + async publish( + @Param('id') id: string, + @Query('newVersion') newVersion: string, + @Body() body: {environmentId: string; releaseNotes: string} + ) { + return this.service.publish(id, newVersion === 'true', body.environmentId, body.releaseNotes) + } + + /** + * @deprecated use workflow trigger instead + */ + @UseGuards(XpertGuard) + @Post(':id/publish/integration') + async publishIntegration(@Param('id') id: string, @Body() integration: Partial) { + return this.commandBus.execute(new XpertPublishIntegrationCommand(id, integration)) + } + + /** + * @deprecated use workflow trigger instead + */ + @UseGuards(XpertGuard) + @Delete(':id/publish/integration/:integration') + async deleteIntegration(@Param('id') id: string, @Param('integration') integration: string,) { + return this.commandBus.execute(new XpertDelIntegrationCommand(id, integration)) + } + + @Get(':id/diagram') + async getDiagram( + @Res() res: Response, + @Param('id') id: string, + @Query('isDraft') isDraft: string, + @Query('agentKey') agentKey: string,) { + try { + const imageData = await this.commandBus.execute(new XpertExportDiagramCommand(id, isDraft === 'true', agentKey)) + res.setHeader('Content-Type', 'image/jpeg') + res.send(Buffer.from(await imageData.arrayBuffer())) + } catch (err) { + console.error(err) + throw new InternalServerErrorException(err.message) + } + } + + @Get(':id/executions') + async getExecutions( + @Param('id') id: string, + @Query('$order', ParseJsonPipe) order?: PaginationParams['order'] + ) { + return this.queryBus.execute(new FindExecutionsByXpertQuery(id, { order })) + } + + @Header('content-type', 'text/event-stream') + @Header('Connection', 'keep-alive') + @Header('Cache-Control', 'no-cache') + @Post(':id/chat') + @Sse() + async chat(@Res() res: Response, + @Param('id') id: string, + @I18nLang() language: LanguagesEnum, + @TimeZone() timeZone: string, + @Body() + body: { + request: TChatRequest + options: TChatOptions + } + ) { + let environment = null + if (body.request.environmentId) { + environment = await this.environmentService.findOne(body.request.environmentId) + } + const observable = await this.enqueueXpertChatTask( + body.request, + { + ...body.options, + xpertId: id, + environment, + language, + timeZone, + from: 'debugger' + } + ) + return observable.pipe( + // Add an operator to send a comment event periodically (30s) to keep the connection alive + keepAlive(30000), + takeUntilClose(res) + ) + } + + @ApiOperation({ summary: 'Delete record' }) + @ApiResponse({ + status: HttpStatus.NO_CONTENT, + description: 'The record has been successfully deleted' + }) + @ApiResponse({ + status: HttpStatus.NOT_FOUND, + description: 'Record not found' + }) + @HttpCode(HttpStatus.ACCEPTED) + @UseGuards(XpertGuard) + @Delete(':id') + async delete( + @Param('id', UUIDValidationPipe) id: string, + ): Promise { + return this.commandBus.execute(new XpertDeleteCommand(id)) + } + + @Get(':id/managers') + async getManagers(@Param('id') id: string) { + const xpert = await this.service.findOne(id, { relations: ['managers'] }) + return xpert.managers.map((u) => new UserPublicDTO(u)) + } + + @Put(':id/managers') + async updateManagers(@Param('id') id: string, @Body() ids: string[]) { + return this.service.updateManagers(id, ids) + } + + @Delete(':id/managers/:userId') + async removeManager(@Param('id') id: string, @Param('userId') userId: string) { + await this.service.removeManager(id, userId) + } + + @Post(':id/managers/bulk/upload') + @UseInterceptors( + FileInterceptor('file', { + storage: new FileStorage().storage({ + dest: path.join('temp'), + prefix: 'managers-csv-upload' + }) + }) + ) + @ApiOperation({ summary: 'Upload and parse CSV file for bulk manager import' }) + @ApiResponse({ + status: HttpStatus.OK, + description: 'Users parsed from CSV file' + }) + async uploadAndParseManagersCsv( + @Param('id') id: string, + @UploadedFileStorage() file: UploadedFile + ): Promise { + if (!file) { + throw new BadRequestException('No file uploaded') + } + + const { path: filePath, mimetype, originalname } = file + + if (mimetype !== 'text/csv' && !originalname.endsWith('.csv')) { + throw new BadRequestException('Only CSV files are supported') + } + + try { + const buffer = await fsPromises.readFile(filePath) + + // Function to check if string contains valid Chinese characters + const containsValidChinese = (str: string): boolean => { + return /[\u4e00-\u9fa5]/.test(str) + } + + // Function to detect if string looks like mis-decoded UTF-8 + const looksLikeMisDecoded = (str: string): boolean => { + return /[´°µÄÂÈøò·¢]{3,}/.test(str) + } + + // Function to check if decoding is likely correct + const isValidDecoding = (decoded: string, encoding: string): boolean => { + if (decoded.includes('\uFFFD')) { + return false + } + + if (encoding === 'utf8' || encoding === 'utf-8') { + try { + Buffer.from(decoded, 'utf8') + return true + } catch { + return false + } + } + + return true + } + + // Try different encodings + const encodingsToTry = ['utf8', 'gbk', 'gb18030', 'gb2312', 'big5'] + let bestDecodedContent = '' + let bestEncoding = 'utf8' + + for (const encoding of encodingsToTry) { + try { + const decoded = iconv.decode(buffer, encoding) + + if (isValidDecoding(decoded, encoding)) { + if (containsValidChinese(decoded) || !looksLikeMisDecoded(decoded)) { + bestDecodedContent = decoded + bestEncoding = encoding + break + } + + if (!bestDecodedContent || decoded.length > bestDecodedContent.length) { + bestDecodedContent = decoded + bestEncoding = encoding + } + } + } catch (error) { + continue + } + } + + // Parse CSV content + const lines = bestDecodedContent.split(/\r?\n/).filter(line => line.trim()) + if (lines.length < 2) { + throw new BadRequestException('CSV file must have at least a header row and one data row') + } + + // Parse header + const headerLine = lines[0] + const headers = this.parseCSVLine(headerLine) + const emailIndex = headers.findIndex(h => h.toLowerCase().trim() === 'email') + + if (emailIndex === -1) { + throw new BadRequestException('CSV file must contain an "email" column') + } + + // Extract emails from CSV + const emails: string[] = [] + for (let i = 1; i < lines.length; i++) { + const values = this.parseCSVLine(lines[i]) + if (values[emailIndex]) { + const email = values[emailIndex].trim() + if (email && email.includes('@')) { + emails.push(email) + } + } + } + + if (emails.length === 0) { + throw new BadRequestException('No valid email addresses found in CSV file') + } + + // Find users by emails + const users = await this.service.findUsersByEmails(emails) + + // Clean up temp file + try { + await fsPromises.unlink(filePath) + } catch { + // Ignore cleanup errors + } + + return users.map((u) => new UserPublicDTO(u)) + } catch (error) { + // Clean up temp file on error + try { + await fsPromises.unlink(filePath) + } catch { + // Ignore cleanup errors + } + + if (error instanceof BadRequestException) { + throw error + } + throw new BadRequestException(`Failed to parse CSV file: ${error.message}`) + } + } + + @Put(':id/managers/bulk') + @ApiOperation({ summary: 'Bulk add managers to xpert' }) + @ApiResponse({ + status: HttpStatus.OK, + description: 'Managers added successfully' + }) + async bulkAddManagers(@Param('id') id: string, @Body() userIds: string[]) { + return this.service.bulkAddManagers(id, userIds) + } + + /** + * Parse a CSV line handling quoted fields + */ + private parseCSVLine(line: string): string[] { + const result: string[] = [] + let current = '' + let inQuotes = false + + for (let i = 0; i < line.length; i++) { + const char = line[i] + + if (char === '"') { + if (inQuotes && line[i + 1] === '"') { + current += '"' + i++ + } else { + inQuotes = !inQuotes + } + } else if (char === ',' && !inQuotes) { + result.push(current) + current = '' + } else { + current += char + } + } + + result.push(current) + return result + } + + @Get(':id/memory') + async getAllMemory(@Param('id') id: string, @Query('types') types: string) { + const _types = types?.split(':').filter((_) => !!_) + return this.service.findAllMemory(id, _types) + } + + @Post(':id/memory/bulk') + async createBulkMemory(@Param('id') id: string, @Body() body: { + type: LongTermMemoryTypeEnum; memories: Array + }) { + return this.service.createBulkMemories(id, body) + } + + /** + * @todo Refactoring is required + */ + @Post(':id/memory/bulk/upload') + @UseInterceptors( + FileInterceptor('file', { + storage: new FileStorage().storage({ + dest: path.join('temp'), // Store in a temporary directory + prefix: 'memory-csv-upload' + }) + }) + ) + @ApiOperation({ summary: 'Upload and parse CSV file for bulk memory import' }) + @ApiResponse({ + status: HttpStatus.OK, + description: 'Memories parsed from CSV file' + }) + async uploadAndParseCsv( + @Param('id') id: string, + @Body() body: { type: LongTermMemoryTypeEnum }, + @UploadedFileStorage() file: UploadedFile + ): Promise> { + const type = body.type + if (!file) { + throw new BadRequestException('No file uploaded') + } + + const { path: filePath, mimetype, originalname } = file + + if (mimetype !== 'text/csv' && !originalname.endsWith('.csv')) { + throw new BadRequestException('Only CSV files are supported') + } + + try { + // Read file buffer + const buffer = await fsPromises.readFile(filePath) + + // Function to check if string contains valid Chinese characters + const containsValidChinese = (str: string): boolean => { + return /[\u4e00-\u9fa5]/.test(str) + } + + // Function to detect if string looks like mis-decoded UTF-8 + const looksLikeMisDecoded = (str: string): boolean => { + return /[´°µÄÂÈøò·¢]{3,}/.test(str) + } + + // Function to check if decoding is likely correct + const isValidDecoding = (decoded: string, encoding: string): boolean => { + if (decoded.includes('\uFFFD')) { + return false + } + + if (encoding === 'utf8' || encoding === 'utf-8') { + try { + Buffer.from(decoded, 'utf8') + return true + } catch { + return false + } + } + + return true + } + + // Try different encodings + const encodingsToTry = ['utf8', 'gbk', 'gb18030', 'gb2312', 'big5'] + + // Check for BOM + let startOffset = 0 + if (buffer[0] === 0xEF && buffer[1] === 0xBB && buffer[2] === 0xBF) { + startOffset = 3 + } + + const bufferToDecode = startOffset > 0 ? buffer.slice(startOffset) : buffer + + let content: string | null = null + let lastError: Error | null = null + + for (const encoding of encodingsToTry) { + try { + const decoded = iconv.decode(bufferToDecode, encoding) + + if (!isValidDecoding(decoded, encoding)) { + continue + } + + if (looksLikeMisDecoded(decoded)) { + continue + } + + if (containsValidChinese(decoded)) { + content = decoded + break + } + + if (encoding === 'utf8' && !content && decoded.length > 0) { + content = decoded + } else if (!content && decoded.length > 0 && !decoded.match(/[^\x00-\x7F]/)) { + content = decoded + } + } catch (err) { + lastError = err + } + } + + if (!content) { + try { + content = iconv.decode(bufferToDecode, 'utf8') + } catch (err) { + throw new BadRequestException(`Failed to decode CSV file. Tried encodings: ${encodingsToTry.join(', ')}. Error: ${lastError?.message || err.message}`) + } + } + + if (!content) { + throw new BadRequestException('Failed to decode CSV file with any known encoding') + } + + // Parse CSV with XLSX + const workbook = XLSX.read(content, { + type: 'string', + codepage: 65001 // UTF-8 codepage + }) + + const sheet = workbook.Sheets[workbook.SheetNames[0]] + const jsonData = XLSX.utils.sheet_to_json(sheet) as any[] + + // Map to memory format based on type + const memories: Array = jsonData.map((row: any) => { + if (type === LongTermMemoryTypeEnum.QA) { + return { + question: row.question || row.问题 || row.問題, + answer: row.answer || row.答案 + } as TMemoryQA + } else if (type === LongTermMemoryTypeEnum.PROFILE) { + return { + profile: row.profile || row.档案 || row.檔案, + context: row.context || row.上下文 + } as TMemoryUserProfile + } + return row + }) + + // Clean up temporary file + try { + await fsPromises.unlink(filePath) + } catch (err) { + // Ignore cleanup errors + } + + return memories + } catch (error) { + // Clean up temporary file on error + try { + await fsPromises.unlink(filePath) + } catch (err) { + // Ignore cleanup errors + } + throw new BadRequestException(`Failed to parse CSV file: ${error.message}`) + } + } + + @Post(':id/memory') + async createMemory(@Param('id') id: string, @Body() body: {type: LongTermMemoryTypeEnum; value: TMemoryQA | TMemoryUserProfile}) { + return this.service.createMemory(id, body) + } + + @Post(':id/memory/search') + async searchMemory(@Param('id') id: string, @Body() body: {type: LongTermMemoryTypeEnum; text: string; isDraft?: boolean;}) { + try { + return await this.queryBus.execute(new SearchXpertMemoryQuery(id, body)) + } catch(err) { + throw new HttpException(getErrorMessage(err), HttpStatus.INTERNAL_SERVER_ERROR) + } + } + + @Delete(':id/memory') + async clearMemory(@Param('id') id: string,) { + try { + return await this.storeService.delete({prefix: Like(`${id}%`)}) + } catch(err) { + throw new HttpException(getErrorMessage(err), HttpStatus.INTERNAL_SERVER_ERROR) + } + } + + @Get(':id/variables') + async getVariables( + @Param('id') id: string, + @Query('environment') environmentId: string, + @Query('inputs') inputs: string + ) { + try { + const inputKeys = inputs ? inputs.split(',').map((value) => value.trim()).filter(Boolean) : undefined + return await this.queryBus.execute( + new XpertAgentVariablesQuery({xpertId: id, isDraft: true, environmentId, inputs: inputKeys}) + ) + } catch (err) { + throw new HttpException(getErrorMessage(err), HttpStatus.INTERNAL_SERVER_ERROR) + } + } + + @Get(':id/agent/:agent/variables') + async getAgentVariables( + @Param('id') id: string, + @Param('agent') agentKey: string, + @Query('environment') environmentId: string, + @Query('isDraft') isDraft: string, + @Query('type') type: 'input' | 'output', + @Query('inputs') inputs: string + ) { + try { + const inputKeys = inputs ? inputs.split(',').map((value) => value.trim()).filter(Boolean) : undefined + return await this.queryBus.execute(new XpertAgentVariablesQuery({ + xpertId: id, type, nodeKey: agentKey, isDraft: !(isDraft === 'false'), environmentId, inputs: inputKeys})) + } catch (err) { + throw new HttpException(getErrorMessage(err), HttpStatus.INTERNAL_SERVER_ERROR) + } + } + + /** + * Get available variables for a workflow node + */ + @Get(':id/workflow/:key/variables') + async getWorkflowVariables( + @Param('id') id: string, + @Param('key') nodeKey: string, + @Query('environment') environmentId: string, + @Query('type') type: 'input' | 'output', + @Query('inputs') inputs: string + ) { + try { + const inputKeys = inputs ? inputs.split(',').map((value) => value.trim()).filter(Boolean) : undefined + return await this.queryBus.execute( + new XpertAgentVariablesQuery({xpertId: id, type, nodeKey, isDraft: true, environmentId, inputs: inputKeys}) + ) + } catch (err) { + throw new HttpException(getErrorMessage(err), HttpStatus.INTERNAL_SERVER_ERROR) + } + } + + @Put(':id/api') + async updateChatApi(@Param('id') id: string, @Body() api: Partial) { + const xpert = await this.service.findOne(id) + await this.service.update(id, {api: {...(xpert.api ?? {}), ...api}}) + if (!api.disabled && !xpert.userId) { + const user = await this.commandBus.execute(new UserCreateCommand({ + username: xpert.slug, + type: UserType.COMMUNICATION, + preferredLanguage: LanguagesEnum.English, + hash: uuidv4(), + })) + await this.service.update(id, {user}) + } + } + + @Put(':id/app') + async updateChatApp(@Param('id') id: string, @Body() app: Partial) { + const xpert = await this.service.findOne(id) + await this.service.update(id, {app: {...(xpert.app ?? {}), ...app}}) + if (app.enabled && !xpert.userId) { + const user = await this.commandBus.execute(new UserCreateCommand({ + username: xpert.slug, + type: UserType.COMMUNICATION, + preferredLanguage: LanguagesEnum.English, + hash: uuidv4(), + })) + await this.service.update(id, {user}) + } + } + + @Post(':id/duplicate') + async duplicate(@Param('id') id: string, @Body() body: {basic: Partial; isDraft: boolean}) { + try { + const xpertDto = await this.commandBus.execute(new XpertExportCommand(id, body.isDraft, false)) + const dsl = instanceToPlain(xpertDto) + return await this.commandBus.execute(new XpertImportCommand({...dsl, team: {...dsl.team, ...body.basic}})) + } catch(err) { + throw new InternalServerErrorException(err.message) + } + } + + // Conversations + + @UseGuards(XpertGuard) + @Get(':id/conversations') + async getConversations( + @Param('id') id: string, + @Query('data', ParseJsonPipe) data: PaginationParams, + @Query('start') start: string, + @Query('end') end: string + ) { + const {where} = data + const result = await this.queryBus.execute(new ChatConversationLogsQuery( + { + ...data, + where: { + ...(where ?? {}), + xpertId: id, + createdAt: start ? Between(new Date(start), new Date(end)) : LessThanOrEqual(new Date(end)) + }, + }, + )) + return { + ...result, + items: result.items.map((item) => new ChatConversationPublicDTO(item)) + } + } + + // Public App + + @Public() + @UseGuards(AnonymousXpertAuthGuard) + @Get(':name/app') + async getChatApp(@Param('name') name: string,) { + const xpert = await this.service.findBySlug(name) + if (!xpert) { + throw new NotFoundException(`Not found xpert '${name}'`) + } + + if (!xpert.app?.enabled) { + throw new ForbiddenException(name) + } + + return new XpertPublicDTO(xpert) + } + + private getPublicUserCondition() { + const userId = RequestContext.currentUserId() + const fromEndUserId = ((RequestContext.currentRequest())).cookies['anonymous.id'] + return userId ? fromEndUserId ? {createdById: userId, fromEndUserId} : {createdById: userId} : {fromEndUserId} + } + + @Public() + @UseGuards(AnonymousXpertAuthGuard) + @Get(':name/conversation/:id') + async getAppConversation(@Param('name') name: string, @Param('id') id: string, + @Query('$relations', ParseJsonPipe) relations?: PaginationParams['relations'], + @Query('$select', ParseJsonPipe) select?: PaginationParams['select'], + ) { + const conversation = await this.queryBus.execute(new GetChatConversationQuery({ + id, + ...this.getPublicUserCondition(), + }, relations)) + return conversation + } + + @Public() + @UseGuards(AnonymousXpertAuthGuard) + @Delete(':name/conversation/:id') + async deleteAppConversation(@Param('name') slug: string, @Param('id') id: string,) { + await this.queryBus.execute(new GetChatConversationQuery({id, ...this.getPublicUserCondition(),})) + await this.commandBus.execute(new ChatConversationDeleteCommand({id, ...this.getPublicUserCondition(),})) + } + + @Public() + @UseGuards(AnonymousXpertAuthGuard) + @Get(':name/conversation') + async getAppConversations(@Param('name') slug: string, + @Query('data', ParseJsonPipe) paginationOptions?: PaginationParams, + ) { + const xpert = await this.service.findBySlug(slug) + const conversation = await this.queryBus.execute(new FindChatConversationQuery({ + ...(paginationOptions.where ?? {}), + ...this.getPublicUserCondition(), + xpertId: xpert.id + }, paginationOptions)) + return conversation + } + + @Public() + @UseGuards(AnonymousXpertAuthGuard) + @Put(':name/conversation/:id') + async updateAppConversation(@Param('name') slug: string, @Param('id') id: string, @Body() entity: Partial) { + await this.queryBus.execute(new FindChatConversationQuery({id, ...this.getPublicUserCondition(),})) + await this.commandBus.execute(new ChatConversationUpsertCommand({ + id, + ...entity + })) + } + + @Public() + @UseGuards(AnonymousXpertAuthGuard) + @Get(':name/conversation/:id/feedbacks') + async getAppFeedbacks(@Param('name') name: string, @Param('id') id: string, + @Query('$relations', ParseJsonPipe) relations?: PaginationParams['relations'], + @Query('$select', ParseJsonPipe) select?: PaginationParams['select'], + ) { + const conversation = await this.queryBus.execute(new FindChatConversationQuery({id, ...this.getPublicUserCondition(),}, {relations})) + return await this.queryBus.execute(new FindMessageFeedbackQuery({conversationId: conversation.id}, relations)) + } + + @Public() + @UseGuards(AnonymousXpertAuthGuard) + @Header('content-type', 'text/event-stream') + @Header('Connection', 'keep-alive') + @Post(':name/chat-app') + @Sse() + async chatApp(@Res() res: Response, @Param('name') name: string, @I18nLang() language: LanguagesEnum, + @Body() body: { request: TChatRequest; options: TChatOptions }) { + const fromEndUserId = ((RequestContext.currentRequest())).cookies['anonymous.id'] + const observable = await this.enqueueXpertChatTask( + body.request, + { + ...body.options, + language, + from: 'webapp', + fromEndUserId + } + ) + return observable.pipe( + // Add an operator to send a comment event periodically (30s) to keep the connection alive + keepAlive(30000), + takeUntilClose(res) + ) + } + + private async enqueueXpertChatTask( + request: TChatRequest, + options: NonNullable[1]> + ) { + const queueTaskId = `xpert-chat-${uuidv4()}` + const sessionKey = request.conversationId ?? options.messageId ?? queueTaskId + + return this.commandBus.execute( + new EnqueueAgentChatMessageCommand( + { + id: queueTaskId, + messageType: AGENT_CHAT_MESSAGE_TYPE, + tenantId: RequestContext.currentTenantId(), + organizationId: RequestContext.getOrganizationId(), + userId: RequestContext.currentUserId(), + sessionKey, + conversationId: request.conversationId, + executionId: options.execution?.id, + source: 'chat', + queueName: XPERT_HANDOFF_QUEUE, + businessKey: sessionKey, + traceId: options.messageId ?? queueTaskId + }, + async () => this.commandBus.execute(new XpertChatCommand(request, options)) + ) + ) + } + + // Statistics + + @UseGuards(RoleGuard) + @Roles(RolesEnum.ADMIN, RolesEnum.SUPER_ADMIN, RolesEnum.TRIAL) + @Get('statistics/xperts') + async getStatisticsXperts(@Query('start') start: string, @Query('end') end: string) { + return await this.queryBus.execute(new StatisticsXpertsQuery(start, end)) + } + + @UseGuards(RoleGuard) + @Roles(RolesEnum.ADMIN, RolesEnum.SUPER_ADMIN, RolesEnum.TRIAL) + @Get('statistics/xpert-conversations') + async getStatisticsXpertConversations(@Query('start') start: string, @Query('end') end: string) { + return await this.queryBus.execute(new StatisticsXpertConversationsQuery(start, end)) + } + + @UseGuards(RoleGuard) + @Roles(RolesEnum.ADMIN, RolesEnum.SUPER_ADMIN, RolesEnum.TRIAL) + @Get('statistics/xpert-messages') + async getStatisticsXpertMessages(@Query('start') start: string, @Query('end') end: string) { + return await this.queryBus.execute(new StatisticsXpertMessagesQuery(start, end)) + } + + @UseGuards(RoleGuard) + @Roles(RolesEnum.ADMIN, RolesEnum.SUPER_ADMIN, RolesEnum.TRIAL) + @Get('statistics/xpert-tokens') + async getStatisticsXpertTokens(@Query('start') start: string, @Query('end') end: string) { + return await this.queryBus.execute(new StatisticsXpertTokensQuery(start, end)) + } + + @UseGuards(RoleGuard) + @Roles(RolesEnum.ADMIN, RolesEnum.SUPER_ADMIN, RolesEnum.TRIAL) + @Get('statistics/xpert-integrations') + async getStatisticsXpertIntegrations(@Query('start') start: string, @Query('end') end: string) { + return await this.queryBus.execute(new StatisticsXpertIntegrationsQuery(start, end)) + } + + @UseGuards(XpertGuard) + @Get(':id/statistics/daily-conversations') + async getDailyConversations(@Param('id') id: string, @Query('start') start: string, @Query('end') end: string) { + return await this.queryBus.execute(new StatisticsDailyConvQuery(start, end, id)) + } + + @UseGuards(XpertGuard) + @Get(':id/statistics/daily-end-users') + async getDailyEndUsers(@Param('id') id: string, @Query('start') start: string, @Query('end') end: string) { + return await this.queryBus.execute(new StatisticsDailyEndUsersQuery(start, end, id)) + } + + @UseGuards(XpertGuard) + @Get(':id/statistics/average-session-interactions') + async getAverageSessionInteractions(@Param('id') id: string, @Query('start') start: string, @Query('end') end: string) { + return await this.queryBus.execute(new StatisticsAverageSessionInteractionsQuery(start, end, id)) + } + + @UseGuards(XpertGuard) + @Get(':id/statistics/daily-messages') + async getDailyMessages(@Param('id') id: string, @Query('start') start: string, @Query('end') end: string) { + return await this.queryBus.execute(new StatisticsDailyMessagesQuery(start, end, id)) + } + + @UseGuards(XpertGuard) + @Get(':id/statistics/tokens-per-second') + async getTokensPerSecond(@Param('id') id: string, @Query('start') start: string, @Query('end') end: string) { + return await this.queryBus.execute(new StatisticsTokensPerSecondQuery(start, end, id)) + } + + @UseGuards(XpertGuard) + @Get(':id/statistics/token-costs') + async getTokenCost(@Param('id') id: string, @Query('start') start: string, @Query('end') end: string) { + return await this.queryBus.execute(new StatisticsTokenCostQuery(start, end, id)) + } + + @UseGuards(XpertGuard) + @Get(':id/statistics/user-satisfaction-rate') + async getUserSatisfactionRate(@Param('id') id: string, @Query('start') start: string, @Query('end') end: string) { + return await this.queryBus.execute(new StatisticsUserSatisfactionRateQuery(start, end, id)) + } + } diff --git a/packages/server-ai/src/xpert/xpert.service.ts b/packages/server-ai/src/xpert/xpert.service.ts index 5a6e03b970..90eb2311d5 100644 --- a/packages/server-ai/src/xpert/xpert.service.ts +++ b/packages/server-ai/src/xpert/xpert.service.ts @@ -1,24 +1,24 @@ import { - ChecklistItem, - convertToUrlPath, - ICopilotStore, - IUser, - IXpertAgentExecution, - LongTermMemoryTypeEnum, - OrderTypeEnum, - TMemoryQA, - TMemoryUserProfile, - TXpertTeamDraft + ChecklistItem, + convertToUrlPath, + ICopilotStore, + IUser, + IXpertAgentExecution, + LongTermMemoryTypeEnum, + OrderTypeEnum, + TMemoryQA, + TMemoryUserProfile, + TXpertTeamDraft } from '@metad/contracts' import { getErrorMessage } from '@metad/server-common' import { - OptionParams, - PaginationParams, - RequestContext, - TenantOrganizationAwareCrudService, - transformWhere, - UserPublicDTO, - UserService + OptionParams, + PaginationParams, + RequestContext, + TenantOrganizationAwareCrudService, + transformWhere, + UserPublicDTO, + UserService } from '@metad/server-core' import { HttpException, HttpStatus, Injectable, NotFoundException } from '@nestjs/common' import { CommandBus, QueryBus } from '@nestjs/cqrs' @@ -40,426 +40,461 @@ import { Xpert } from './xpert.entity' @Injectable() export class XpertService extends TenantOrganizationAwareCrudService { - constructor( - @InjectRepository(Xpert) - public readonly repository: Repository, - private readonly storeService: CopilotStoreService, - private readonly userService: UserService, - private readonly commandBus: CommandBus, - private readonly queryBus: QueryBus, - private readonly eventEmitter: EventEmitter2, - private readonly triggerRegistry: WorkflowTriggerRegistry, - private readonly sandboxService: SandboxService - ) { - super(repository) - } - - /** - * To solve the problem that Update cannot create OneToOne relation, it is uncertain whether using save to update might pose risks - */ - async update(id: string, entity: Partial) { - const _entity = await super.findOne(id) - assign(_entity, entity) - return await this.repository.save(_entity) - } - - /** - * Verify the uniqueness of the slug generated by Name in the system to ensure that it is unique across the entire database - * - * @param name - * @returns - */ - async validateName(name: string) { - const slug = convertToUrlPath(name) - if (slug.length < 5) { - return false - } - const count = await this.repository.count({ - where: { - slug, - latest: true - } - }) - - return !count - } - - async getAllByWorkspace(workspaceId: string, data: PaginationParams, published: boolean, user: IUser) { - const { select, relations, order, take } = data ?? {} - let { where } = data ?? {} - where = transformWhere(where ?? {}) - if (workspaceId === 'null' || workspaceId === 'undefined' || !workspaceId) { - where = { - ...(>where), - workspaceId: IsNull(), - createdById: user.id - } - } else { - const workspace = await this.queryBus.execute(new GetXpertWorkspaceQuery(user, { id: workspaceId })) - if (!workspace) { - throw new NotFoundException(`Not found or no auth for xpert workspace '${workspaceId}'`) - } - - where = { - ...(>where), - workspaceId: workspaceId - } - } - if (published) { - where.version = Not(IsNull()) - } - - return this.findAll({ - select, - where, - relations, - order, - take - }) - } - - async countMy(where: FindOptionsWhere) { - const userId = RequestContext.currentUserId() - const { items: userWorkspaces } = await this.queryBus.execute(new MyXpertWorkspaceQuery(userId, {})) - - where = { - ...(>where ?? {}), - publishAt: Not(IsNull()), - createdById: userId - } - - const xpertsCreatedByUser = await this.findAll({ - select: ['id'], - where, - }) - - const baseQuery = this.repository - .createQueryBuilder('xpert') - .innerJoin('xpert.managers', 'manager', 'manager.id = :userId', { userId }) - - const xpertsManagedByUser = await baseQuery - .where({ - ...(where ?? {}), - publishAt: Not(IsNull()), - tenantId: RequestContext.currentTenantId(), - organizationId: RequestContext.getOrganizationId() - }) - .select(['xpert.id']) - .getMany() - - const xpertsInUserWorkspaces = await this.repository.find({ - where: { - ...(where ?? {}), - publishAt: Not(IsNull()), - workspaceId: In(userWorkspaces.map((workspace) => workspace.id)) - }, - select: ['id'] - }) - - const allXperts = uniqBy( - [...xpertsCreatedByUser.items, ...xpertsManagedByUser, ...xpertsInUserWorkspaces], - 'id' - ) - - return allXperts.length - } - - async getMyAll(params: PaginationParams) { - const userId = RequestContext.currentUserId() - const { items: userWorkspaces } = await this.queryBus.execute(new MyXpertWorkspaceQuery(userId, {})) - - const { relations, order, take } = params ?? {} - let { where } = params ?? {} - where = where ?? {} - - where = { - ...(>where), - publishAt: Not(IsNull()), - createdById: userId - } - - const xpertsCreatedByUser = await this.findAll({ - where, - relations, - order, - take - }) - - const baseQuery = this.repository - .createQueryBuilder('xpert') - .innerJoin('xpert.managers', 'manager', 'manager.id = :userId', { userId }) - // add relations - relations?.forEach((relation) => baseQuery.leftJoinAndSelect('xpert.' + relation, relation)) - if (order) { - Object.keys(order).forEach((name) => { - baseQuery.addOrderBy(`xpert.${name}`, order[name]) - }) - } - const xpertsManagedByUser = await baseQuery - .where({ - ...(params.where ?? {}), - publishAt: Not(IsNull()), - tenantId: RequestContext.currentTenantId(), - organizationId: RequestContext.getOrganizationId() - }) - .take(take) - .getMany() - - const xpertsInUserWorkspaces = await this.repository.find({ - where: { - ...(params.where ?? {}), - publishAt: Not(IsNull()), - workspaceId: In(userWorkspaces.map((workspace) => workspace.id)) - }, - relations, - order, - take - }) - - const allXperts = uniqBy( - [...xpertsCreatedByUser.items, ...xpertsManagedByUser, ...xpertsInUserWorkspaces], - 'id' - ) - - return { - items: allXperts.map((item) => new XpertIdentiDto(item)), - total: allXperts.length - } - } - - async getTeam(id: string, options?: OptionParams) { - const { relations } = options ?? {} - const team = await this.findOne(id, { - relations: uniq([...(relations ?? []), 'agents', 'toolsets', 'knowledgebases']) - }) - return team - } - - async save(entity: Xpert) { - return await this.repository.save(entity) - } - - async saveDraft(id: string, draft: TXpertTeamDraft) { - const xpert = await this.findOne(id) - xpert.draft = { - ...draft, - team: { - ...draft.team, - updatedAt: new Date(), - updatedById: RequestContext.currentUserId() - } - } as TXpertTeamDraft - - xpert.draft.checklist = await this.validate(xpert.draft) - - await this.repository.save(xpert) - return xpert.draft - } - - async updateDraft(id: string, draft: TXpertTeamDraft) { - const xpert = await this.findOne(id) - xpert.draft = { - ...(xpert.draft ?? {}), - ...draft, - team: { - ...(xpert.draft?.team ?? {}), - ...(draft.team ?? {}), - updatedAt: new Date(), - updatedById: RequestContext.currentUserId() - } - } as TXpertTeamDraft - - xpert.draft.checklist = await this.validate(xpert.draft) - - await this.repository.save(xpert) - return xpert.draft - } - - async validate(draft: TXpertTeamDraft) { - const freeNodeValidator = new FreeNodeValidator() - - const results: ChecklistItem[] = [] - - const res = await freeNodeValidator.validate(draft) - results.push(...res) - - // More validators events - const validators = await this.eventEmitter.emitAsync(EventNameXpertValidate, new XpertDraftValidateEvent(draft)) - validators.forEach((items) => { - if (items) { - results.push(...items) - } - }) - return results - } - - async publish(id: string, newVersion: boolean, environmentId: string, notes: string) { - return await this.commandBus.execute(new XpertPublishCommand(id, newVersion, environmentId, notes)) - } - - async allVersions(id: string) { - const xpert = await this.findOne(id) - const { items: allVersions } = await this.findAll({ - where: { - workspaceId: xpert.workspaceId ?? IsNull(), - type: xpert.type, - slug: xpert.slug - } - }) - - return allVersions.map((item) => ({ - id: item.id, - version: item.version, - latest: item.latest, - publishAt: item.publishAt, - releaseNotes: item.releaseNotes - })) - } - - async setAsLatest(id: string) { - const xpert = await this.findOne(id) - if (!xpert.latest) { - const { items: xperts } = await this.findAll({ - where: { - workspaceId: xpert.workspaceId ?? IsNull(), - type: xpert.type, - slug: xpert.slug, - latest: true - } - }) - - xperts.forEach((item) => (item.latest = false)) - xpert.latest = true - await this.repository.save([...xperts, xpert]) - } - } - - async deleteXpert(id: string) { - const xpert = await this.findOne(id) - - if (xpert.latest) { - // Delete all versions if it is latest version - return await this.softDelete({ name: xpert.name, deletedAt: IsNull() }) - } else { - // Delete current version team - return await this.softDelete(xpert.id) - } - } - - async updateManagers(id: string, ids: string[]) { - const xpert = await this.findOne(id, { relations: ['managers'] }) - const { items } = await this.userService.findAll({ where: { id: In(ids) } }) - xpert.managers = items - await this.repository.save(xpert) - return xpert.managers.map((u) => new UserPublicDTO(u)) - } - - async removeManager(id: string, userId: string) { - const xpert = await this.findOne(id, { relations: ['managers'] }) - if (!xpert) { - throw new NotFoundException(`Xpert with id ${id} not found`) - } - - const managerIndex = xpert.managers.findIndex((manager) => manager.id === userId) - if (managerIndex === -1) { - throw new NotFoundException(`Manager with id ${userId} not found in Xpert ${id}`) - } - - xpert.managers.splice(managerIndex, 1) - await this.repository.save(xpert) - } - - async findBySlug(slug: string, relations?: string[]) { - return await this.repository.findOne({ - where: { - slug, - latest: true, - publishAt: Not(IsNull()) - }, - relations: uniq((relations ?? []).concat(['user', 'createdBy', 'organization'])) - }) - } - - async createMemory(xpertId: string, body: { type: LongTermMemoryTypeEnum; value: TMemoryQA | TMemoryUserProfile }) { - const xpert = await this.findOne(xpertId, { relations: ['agent'] }) - const memory = xpert.memory - const tenantId = RequestContext.currentTenantId() - const organizationId = RequestContext.getOrganizationId() - const execution: IXpertAgentExecution = {} - const embeddings = await this.queryBus.execute( - new GetXpertMemoryEmbeddingsQuery(tenantId, organizationId, memory, { - tokenCallback: (token) => { - execution.embedTokens += token ?? 0 - } - }) - ) - - await this.commandBus.execute( - new CopilotStoreBulkPutCommand( - body.type, - [body.value], - [xpertId, body.type || LongTermMemoryTypeEnum.QA], - embeddings - ) - ) - } - - async createBulkMemories( - xpertId: string, - body: { type: LongTermMemoryTypeEnum; memories: Array } - ) { - const xpert = await this.findOne(xpertId, { relations: ['agent'] }) - const memory = xpert.memory - const tenantId = RequestContext.currentTenantId() - const organizationId = RequestContext.getOrganizationId() - const execution: IXpertAgentExecution = {} - const embeddings = await this.queryBus.execute( - new GetXpertMemoryEmbeddingsQuery(tenantId, organizationId, memory, { - tokenCallback: (token) => { - execution.embedTokens += token ?? 0 - } - }) - ) - await this.commandBus.execute( - new CopilotStoreBulkPutCommand( - body.type, - body.memories, - [xpertId, body.type || LongTermMemoryTypeEnum.QA], - embeddings - ) - ) - } - - async findAllMemory(id: string, types: string[]) { - const where = {} as FindOptionsWhere - const _types = types - if (_types?.length > 1) { - where.prefix = In(_types.map((type) => `${id}${type ? `:${type}` : ''}`)) - } else if (_types?.length === 1) { - const type = _types[0] - where.prefix = `${id}${type ? `:${type}` : ''}` - } else { - where.prefix = Like(`${id}%`) - } - - try { - return await this.storeService.findAll({ - where, - relations: ['createdBy'], - order: { createdAt: OrderTypeEnum.DESC } - }) - } catch (err) { - throw new HttpException(getErrorMessage(err), HttpStatus.INTERNAL_SERVER_ERROR) - } - } - - async getTriggerProviders() { - return this.triggerRegistry.list().map((provider) => ({ - ...provider.meta - })) - } - - async getSandboxProviders() { - return this.sandboxService.listProviders() - } + constructor( + @InjectRepository(Xpert) + public readonly repository: Repository, + private readonly storeService: CopilotStoreService, + private readonly userService: UserService, + private readonly commandBus: CommandBus, + private readonly queryBus: QueryBus, + private readonly eventEmitter: EventEmitter2, + private readonly triggerRegistry: WorkflowTriggerRegistry, + private readonly sandboxService: SandboxService + ) { + super(repository) + } + + /** + * To solve the problem that Update cannot create OneToOne relation, it is uncertain whether using save to update might pose risks + */ + async update(id: string, entity: Partial) { + const _entity = await super.findOne(id) + assign(_entity, entity) + return await this.repository.save(_entity) + } + + /** + * Verify the uniqueness of the slug generated by Name in the system to ensure that it is unique across the entire database + * + * @param name + * @returns + */ + async validateName(name: string) { + const slug = convertToUrlPath(name) + if (slug.length < 5) { + return false + } + const count = await this.repository.count({ + where: { + slug, + latest: true + } + }) + + return !count + } + + async getAllByWorkspace(workspaceId: string, data: PaginationParams, published: boolean, user: IUser) { + const { select, relations, order, take } = data ?? {} + let { where } = data ?? {} + where = transformWhere(where ?? {}) + if (workspaceId === 'null' || workspaceId === 'undefined' || !workspaceId) { + where = { + ...(>where), + workspaceId: IsNull(), + createdById: user.id + } + } else { + const workspace = await this.queryBus.execute(new GetXpertWorkspaceQuery(user, { id: workspaceId })) + if (!workspace) { + throw new NotFoundException(`Not found or no auth for xpert workspace '${workspaceId}'`) + } + + where = { + ...(>where), + workspaceId: workspaceId + } + } + if (published) { + where.version = Not(IsNull()) + } + + return this.findAll({ + select, + where, + relations, + order, + take + }) + } + + async countMy(where: FindOptionsWhere) { + const userId = RequestContext.currentUserId() + const { items: userWorkspaces } = await this.queryBus.execute(new MyXpertWorkspaceQuery(userId, {})) + + where = { + ...(>where ?? {}), + publishAt: Not(IsNull()), + createdById: userId + } + + const xpertsCreatedByUser = await this.findAll({ + select: ['id'], + where, + }) + + const baseQuery = this.repository + .createQueryBuilder('xpert') + .innerJoin('xpert.managers', 'manager', 'manager.id = :userId', { userId }) + + const xpertsManagedByUser = await baseQuery + .where({ + ...(where ?? {}), + publishAt: Not(IsNull()), + tenantId: RequestContext.currentTenantId(), + organizationId: RequestContext.getOrganizationId() + }) + .select(['xpert.id']) + .getMany() + + const xpertsInUserWorkspaces = await this.repository.find({ + where: { + ...(where ?? {}), + publishAt: Not(IsNull()), + workspaceId: In(userWorkspaces.map((workspace) => workspace.id)) + }, + select: ['id'] + }) + + const allXperts = uniqBy( + [...xpertsCreatedByUser.items, ...xpertsManagedByUser, ...xpertsInUserWorkspaces], + 'id' + ) + + return allXperts.length + } + + async getMyAll(params: PaginationParams) { + const userId = RequestContext.currentUserId() + const { items: userWorkspaces } = await this.queryBus.execute(new MyXpertWorkspaceQuery(userId, {})) + + const { relations, order, take } = params ?? {} + let { where } = params ?? {} + where = where ?? {} + + where = { + ...(>where), + publishAt: Not(IsNull()), + createdById: userId + } + + const xpertsCreatedByUser = await this.findAll({ + where, + relations, + order, + take + }) + + const baseQuery = this.repository + .createQueryBuilder('xpert') + .innerJoin('xpert.managers', 'manager', 'manager.id = :userId', { userId }) + // add relations + relations?.forEach((relation) => baseQuery.leftJoinAndSelect('xpert.' + relation, relation)) + if (order) { + Object.keys(order).forEach((name) => { + baseQuery.addOrderBy(`xpert.${name}`, order[name]) + }) + } + const xpertsManagedByUser = await baseQuery + .where({ + ...(params.where ?? {}), + publishAt: Not(IsNull()), + tenantId: RequestContext.currentTenantId(), + organizationId: RequestContext.getOrganizationId() + }) + .take(take) + .getMany() + + const xpertsInUserWorkspaces = await this.repository.find({ + where: { + ...(params.where ?? {}), + publishAt: Not(IsNull()), + workspaceId: In(userWorkspaces.map((workspace) => workspace.id)) + }, + relations, + order, + take + }) + + const allXperts = uniqBy( + [...xpertsCreatedByUser.items, ...xpertsManagedByUser, ...xpertsInUserWorkspaces], + 'id' + ) + + return { + items: allXperts.map((item) => new XpertIdentiDto(item)), + total: allXperts.length + } + } + + async getTeam(id: string, options?: OptionParams) { + const { relations } = options ?? {} + const team = await this.findOne(id, { + relations: uniq([...(relations ?? []), 'agents', 'toolsets', 'knowledgebases']) + }) + return team + } + + async save(entity: Xpert) { + return await this.repository.save(entity) + } + + async saveDraft(id: string, draft: TXpertTeamDraft) { + const xpert = await this.findOne(id) + xpert.draft = { + ...draft, + team: { + ...draft.team, + updatedAt: new Date(), + updatedById: RequestContext.currentUserId() + } + } as TXpertTeamDraft + + xpert.draft.checklist = await this.validate(xpert.draft) + + await this.repository.save(xpert) + return xpert.draft + } + + async updateDraft(id: string, draft: TXpertTeamDraft) { + const xpert = await this.findOne(id) + xpert.draft = { + ...(xpert.draft ?? {}), + ...draft, + team: { + ...(xpert.draft?.team ?? {}), + ...(draft.team ?? {}), + updatedAt: new Date(), + updatedById: RequestContext.currentUserId() + } + } as TXpertTeamDraft + + xpert.draft.checklist = await this.validate(xpert.draft) + + await this.repository.save(xpert) + return xpert.draft + } + + async validate(draft: TXpertTeamDraft) { + const freeNodeValidator = new FreeNodeValidator() + + const results: ChecklistItem[] = [] + + const res = await freeNodeValidator.validate(draft) + results.push(...res) + + // More validators events + const validators = await this.eventEmitter.emitAsync(EventNameXpertValidate, new XpertDraftValidateEvent(draft)) + validators.forEach((items) => { + if (items) { + results.push(...items) + } + }) + return results + } + + async publish(id: string, newVersion: boolean, environmentId: string, notes: string) { + return await this.commandBus.execute(new XpertPublishCommand(id, newVersion, environmentId, notes)) + } + + async allVersions(id: string) { + const xpert = await this.findOne(id) + const { items: allVersions } = await this.findAll({ + where: { + workspaceId: xpert.workspaceId ?? IsNull(), + type: xpert.type, + slug: xpert.slug + } + }) + + return allVersions.map((item) => ({ + id: item.id, + version: item.version, + latest: item.latest, + publishAt: item.publishAt, + releaseNotes: item.releaseNotes + })) + } + + async setAsLatest(id: string) { + const xpert = await this.findOne(id) + if (!xpert.latest) { + const { items: xperts } = await this.findAll({ + where: { + workspaceId: xpert.workspaceId ?? IsNull(), + type: xpert.type, + slug: xpert.slug, + latest: true + } + }) + + xperts.forEach((item) => (item.latest = false)) + xpert.latest = true + await this.repository.save([...xperts, xpert]) + } + } + + async deleteXpert(id: string) { + const xpert = await this.findOne(id) + + if (xpert.latest) { + // Delete all versions if it is latest version + return await this.softDelete({ name: xpert.name, deletedAt: IsNull() }) + } else { + // Delete current version team + return await this.softDelete(xpert.id) + } + } + + async updateManagers(id: string, ids: string[]) { + const xpert = await this.findOne(id, { relations: ['managers'] }) + const { items } = await this.userService.findAll({ where: { id: In(ids) } }) + xpert.managers = items + await this.repository.save(xpert) + return xpert.managers.map((u) => new UserPublicDTO(u)) + } + + async removeManager(id: string, userId: string) { + const xpert = await this.findOne(id, { relations: ['managers'] }) + if (!xpert) { + throw new NotFoundException(`Xpert with id ${id} not found`) + } + + const managerIndex = xpert.managers.findIndex((manager) => manager.id === userId) + if (managerIndex === -1) { + throw new NotFoundException(`Manager with id ${userId} not found in Xpert ${id}`) + } + + xpert.managers.splice(managerIndex, 1) + await this.repository.save(xpert) + } + + async findUsersByEmails(emails: string[]): Promise { + const { items } = await this.userService.findAll({ + where: { email: In(emails) } + }) + return items + } + + async bulkAddManagers(id: string, userIds: string[]) { + const xpert = await this.findOne(id, { relations: ['managers'] }) + if (!xpert) { + throw new NotFoundException(`Xpert with id ${id} not found`) + } + + // Get existing manager IDs + const existingManagerIds = new Set(xpert.managers.map((m) => m.id)) + + // Filter out users that are already managers + const newUserIds = userIds.filter((userId) => !existingManagerIds.has(userId)) + + if (newUserIds.length === 0) { + return xpert.managers.map((u) => new UserPublicDTO(u)) + } + + // Find new users to add + const { items: newUsers } = await this.userService.findAll({ + where: { id: In(newUserIds) } + }) + + // Add new managers + xpert.managers = [...xpert.managers, ...newUsers] + await this.repository.save(xpert) + + return xpert.managers.map((u) => new UserPublicDTO(u)) + } + + async findBySlug(slug: string, relations?: string[]) { + return await this.repository.findOne({ + where: { + slug, + latest: true, + publishAt: Not(IsNull()) + }, + relations: uniq((relations ?? []).concat(['user', 'createdBy', 'organization'])) + }) + } + + async createMemory(xpertId: string, body: { type: LongTermMemoryTypeEnum; value: TMemoryQA | TMemoryUserProfile }) { + const xpert = await this.findOne(xpertId, { relations: ['agent'] }) + const memory = xpert.memory + const tenantId = RequestContext.currentTenantId() + const organizationId = RequestContext.getOrganizationId() + const execution: IXpertAgentExecution = {} + const embeddings = await this.queryBus.execute( + new GetXpertMemoryEmbeddingsQuery(tenantId, organizationId, memory, { + tokenCallback: (token) => { + execution.embedTokens += token ?? 0 + } + }) + ) + + await this.commandBus.execute( + new CopilotStoreBulkPutCommand( + body.type, + [body.value], + [xpertId, body.type || LongTermMemoryTypeEnum.QA], + embeddings + ) + ) + } + + async createBulkMemories( + xpertId: string, + body: { type: LongTermMemoryTypeEnum; memories: Array } + ) { + const xpert = await this.findOne(xpertId, { relations: ['agent'] }) + const memory = xpert.memory + const tenantId = RequestContext.currentTenantId() + const organizationId = RequestContext.getOrganizationId() + const execution: IXpertAgentExecution = {} + const embeddings = await this.queryBus.execute( + new GetXpertMemoryEmbeddingsQuery(tenantId, organizationId, memory, { + tokenCallback: (token) => { + execution.embedTokens += token ?? 0 + } + }) + ) + await this.commandBus.execute( + new CopilotStoreBulkPutCommand( + body.type, + body.memories, + [xpertId, body.type || LongTermMemoryTypeEnum.QA], + embeddings + ) + ) + } + + async findAllMemory(id: string, types: string[]) { + const where = {} as FindOptionsWhere + const _types = types + if (_types?.length > 1) { + where.prefix = In(_types.map((type) => `${id}${type ? `:${type}` : ''}`)) + } else if (_types?.length === 1) { + const type = _types[0] + where.prefix = `${id}${type ? `:${type}` : ''}` + } else { + where.prefix = Like(`${id}%`) + } + + try { + return await this.storeService.findAll({ + where, + relations: ['createdBy'], + order: { createdAt: OrderTypeEnum.DESC } + }) + } catch (err) { + throw new HttpException(getErrorMessage(err), HttpStatus.INTERNAL_SERVER_ERROR) + } + } + + async getTriggerProviders() { + return this.triggerRegistry.list().map((provider) => ({ + ...provider.meta + })) + } + + async getSandboxProviders() { + return this.sandboxService.listProviders() + } }