Feat/aggregate node

packages/api/src/core/ai.ts

@@ -45,19 +45,29 @@ export async function generateAiCode(
   return response.data
 }
 
+export interface AiAggregateRequest {
+  prompt: string
+  fields?: Array<{
+    name: string
+    type: string
+    primaryKey?: boolean
+    nullable?: boolean
+    comment?: string
+  }>
+  existingCode?: string
+}
+
 /**
- * Generate JavaScript code using AI with SSE streaming
- * @param data - The request data containing prompt and optional fields
- * @param callbacks - Callbacks for handling SSE events
- * @returns AbortController to cancel the request
+ * Internal helper: create an SSE streaming request
  */
-export function generateAiCodeStream(
-  data: AiGenerateRequest,
+function createSSEStream(
+  url: string,
+  data: any,
   callbacks: SSECallbacks,
 ): AbortController {
   const controller = new AbortController()
 
-  fetch(`${AI_BASE_URL}/generate/stream`, {
+  fetch(url, {
     method: 'POST',
     headers: {
       'Content-Type': 'application/json',
@@ -90,9 +100,8 @@ export function generateAiCodeStream(
 
         buffer += decoder.decode(value, { stream: true })
 
-        // Parse SSE events from buffer
         const lines = buffer.split('\n')
-        buffer = lines.pop() || '' // Keep incomplete line in buffer
+        buffer = lines.pop() || ''
 
         let currentEvent = ''
         let currentData = ''
@@ -103,7 +112,6 @@ export function generateAiCodeStream(
           } else if (line.startsWith('data: ')) {
             currentData = line.slice(6)
           } else if (line === '' && currentEvent && currentData) {
-            // Empty line signals end of event
             try {
               const parsed = JSON.parse(currentData)
               if (currentEvent === 'chunk' && parsed.content) {
@@ -130,3 +138,29 @@ export function generateAiCodeStream(
 
   return controller
 }
+
+/**
+ * Generate MongoDB aggregation pipeline using AI with SSE streaming
+ * @param data - The request data containing prompt, fields and optional existing pipeline
+ * @param callbacks - Callbacks for handling SSE events
+ * @returns AbortController to cancel the request
+ */
+export function generateAiAggregateStream(
+  data: AiAggregateRequest,
+  callbacks: SSECallbacks,
+): AbortController {
+  return createSSEStream(`${AI_BASE_URL}/aggregate/stream`, data, callbacks)
+}
+
+/**
+ * Generate JavaScript code using AI with SSE streaming
+ * @param data - The request data containing prompt and optional fields
+ * @param callbacks - Callbacks for handling SSE events
+ * @returns AbortController to cancel the request
+ */
+export function generateAiCodeStream(
+  data: AiGenerateRequest,
+  callbacks: SSECallbacks,
+): AbortController {
+  return createSSEStream(`${AI_BASE_URL}/generate/stream`, data, callbacks)
+}

packages/business/src/components/logs/NodeLog.vue

@@ -420,6 +420,7 @@ function addLogTagsFilter(params: any) {
     if (
       node &&
       [
+        'custom_processor',
         'js_processor',
         'migrate_js_processor',
         'standard_js_processor',

packages/dag/src/EditorView.vue

@@ -82,8 +82,8 @@ const init = async () => {
   await dataflowStore.initPdkProperties()
 
   if (taskId) {
+    await initNodeType()
     await dataflowStore.fetchDataflow(taskId)
-    await initNodeType(dataflowStore.dataflow.syncType)
     // nextTick(() => {
     //   setTimeout(() => {
     //     canvasRef.value.fitViewWithOffset({ duration: 0, maxZoom: 1 })
@@ -100,7 +100,7 @@ const init = async () => {
       syncType = 'migrate'
       targetRoute = 'MigrateEditor'
     }
-    await initNodeType(syncType!)
+    await initNodeType()
     await dataflowStore.createDataflow(syncType)
     router.push({
       name: targetRoute,

packages/dag/src/MonitorView.vue

@@ -10,7 +10,7 @@ import { TextEditable } from '@tap/component/src/base/text-editable'
 import Time from '@tap/shared/src/time'
 import { useDark } from '@vueuse/core'
 import { debounce } from 'lodash-es'
-import { computed, onUnmounted, provide, ref, watch } from 'vue'
+import { computed, onUnmounted, provide, ref } from 'vue'
 import { useRoute, useRouter } from 'vue-router'
 import Canvas from './Canvas.vue'
 import ConsolePanel from './components/migration/ConsolePanel.vue'
@@ -561,13 +561,6 @@ const initMonitor = debounce(() => {
   startLoadData()
 }, 200)
 
-watch(
-  () => dataflowStore.stateIsReadonly,
-  (v) => {
-    console.trace('stateIsReadonly', v)
-  },
-)
-
 function handleOpenDetail(node: any) {
   if (['mem_cache'].includes(node.type)) return
   nodeDetailDialogId.value = node.id
@@ -599,8 +592,8 @@ const init = async () => {
   await dataflowStore.initPdkProperties()
 
   if (taskId) {
+    await initNodeType()
     await dataflowStore.fetchDataflow(taskId)
-    await initNodeType(dataflowStore.dataflow.syncType)
   }
   initMonitor()
   initWS()

packages/dag/src/composables/useCanvasOperation.ts

@@ -113,7 +113,12 @@ export function useCanvasOperation() {
   )
 
   const isSyncTask = computed(() => {
-    return ['DataflowNew', 'DataflowEditor'].includes(route.name)
+    return [
+      'DataflowNew',
+      'DataflowEditor',
+      'TaskMonitor',
+      'MigrationMonitorViewer', // 任务记录也加载自定节点
+    ].includes(route.name)
   })
 
   const monitorRoute = computed(() => {
@@ -243,10 +248,10 @@ export function useCanvasOperation() {
     } */,
   ]
 
-  const initNodeType = async (syncType: string) => {
-    let nodes = syncType === 'sync' ? syncProcessor : migrateProcessor
+  const initNodeType = async () => {
+    let nodes = isSyncTask.value ? syncProcessor : migrateProcessor
     //仅企业版有的节点
-    if (isDaas && syncType === 'sync') {
+    if (isDaas && isSyncTask.value) {
       const isDaasNode = [
         {
           name: t('packages_dag_src_editor_join'),
@@ -258,7 +263,7 @@ export function useCanvasOperation() {
     dataflowStore.addProcessorNode(nodes.filter((item) => !item.hidden))
     // dataflowStore.addResourceIns(allResourceIns)
 
-    if (syncType === 'sync' && hasFeature('customProcessor')) {
+    if (isSyncTask.value && hasFeature('customProcessor')) {
       await dataflowStore.loadCustomNode()
     }
   }

packages/form/package.json

@@ -32,6 +32,7 @@
     "element-plus": "catalog:",
     "highlight.js": "^11.9.0",
     "lodash": "catalog:",
+    "monaco-editor": "catalog:",
     "resize-observer-polyfill": "catalog:",
     "tiny-emitter": "catalog:",
     "vue": "catalog:",

packages/form/src/components/aggregate/AggregateFields.tsx

@@ -0,0 +1,167 @@
+import { useI18n } from '@tap/i18n'
+import { defineComponent, type PropType } from 'vue'
+import { BaseFieldSelect } from '../field-select'
+
+export interface AggregateField {
+  id: string
+  outputField: string
+  operator: string
+  sourceField: string
+}
+
+const AGG_OPERATORS = [
+  { label: '$sum', value: '$sum' },
+  { label: '$avg', value: '$avg' },
+  { label: '$min', value: '$min' },
+  { label: '$max', value: '$max' },
+  { label: '$count', value: '$count' },
+  { label: '$first', value: '$first' },
+  { label: '$last', value: '$last' },
+  { label: '$push', value: '$push' },
+  { label: '$addToSet', value: '$addToSet' },
+]
+
+let aggIdCounter = 0
+function genId() {
+  return `agg_${++aggIdCounter}_${Date.now()}`
+}
+
+export const AggregateFields = defineComponent({
+  name: 'AggregateFields',
+  props: {
+    disabled: Boolean,
+    fields: {
+      type: Array as PropType<AggregateField[]>,
+      default: () => [],
+    },
+    fieldOptions: {
+      type: Array as PropType<any[]>,
+      default: () => [],
+    },
+    loading: Boolean,
+  },
+  emits: ['update:fields'],
+  setup(props, { emit }) {
+    const { t } = useI18n()
+
+    const addField = () => {
+      const newField: AggregateField = {
+        id: genId(),
+        outputField: '',
+        operator: '$sum',
+        sourceField: '',
+      }
+      emit('update:fields', [...props.fields, newField])
+    }
+
+    const removeField = (index: number) => {
+      const next = [...props.fields]
+      next.splice(index, 1)
+      emit('update:fields', next)
+    }
+
+    const updateField = (index: number, patch: Partial<AggregateField>) => {
+      const next = props.fields.map((f, i) =>
+        i === index ? { ...f, ...patch } : f,
+      )
+      emit('update:fields', next)
+    }
+
+    const onDragStart = (e: DragEvent, index: number) => {
+      e.dataTransfer?.setData('text/plain', String(index))
+    }
+
+    const onDrop = (e: DragEvent, toIndex: number) => {
+      e.preventDefault()
+      const fromIndex = Number(e.dataTransfer?.getData('text/plain'))
+      if (isNaN(fromIndex) || fromIndex === toIndex) return
+      const next = [...props.fields]
+      const [moved] = next.splice(fromIndex, 1)
+      next.splice(toIndex, 0, moved!)
+      emit('update:fields', next)
+    }
+
+    return () => (
+      <div class="aggregate-fields">
+        {props.fields.map((af, index) => (
+          <div
+            key={af.id}
+            class="aggregate-fields__item"
+            draggable
+            onDragstart={(e: DragEvent) => onDragStart(e, index)}
+            onDragover={(e: DragEvent) => e.preventDefault()}
+            onDrop={(e: DragEvent) => onDrop(e, index)}
+          >
+            <div class="aggregate-fields__drag-handle">
+              <el-icon size={14}>
+                <i-lucide-grip-vertical />
+              </el-icon>
+            </div>
+
+            <ElInput
+              disabled={props.disabled}
+              class="aggregate-fields__output"
+              modelValue={af.outputField}
+              onUpdate:modelValue={(val: string) =>
+                updateField(index, { outputField: val })
+              }
+              placeholder={t('packages_form_aggregate_output_field')}
+            />
+
+            <ElSelect
+              disabled={props.disabled}
+              class="aggregate-fields__operator"
+              modelValue={af.operator}
+              onUpdate:modelValue={(val: string) =>
+                updateField(index, { operator: val })
+              }
+              style={{ width: '120px' }}
+            >
+              {AGG_OPERATORS.map((op) => (
+                <ElOption key={op.value} label={op.label} value={op.value} />
+              ))}
+            </ElSelect>
+
+            <BaseFieldSelect
+              disabled={props.disabled}
+              class="aggregate-fields__source"
+              modelValue={af.sourceField}
+              options={props.fieldOptions}
+              loading={props.loading}
+              {...({
+                filterable: true,
+                placeholder: t('packages_form_aggregate_select_source_field'),
+                disabled: af.operator === '$count',
+                onChange: (val: string) =>
+                  updateField(index, { sourceField: val }),
+              } as any)}
+            />
+
+            <el-button
+              disabled={props.disabled}
+              class="aggregate-fields__delete"
+              text
+              type="danger"
+              onClick={() => removeField(index)}
+              icon={IconLucideTrash2}
+              size="small"
+            />
+          </div>
+        ))}
+
+        <ElButton
+          disabled={props.disabled}
+          type="primary"
+          text
+          size="small"
+          onClick={addField}
+        >
+          <el-icon class="mr-1">
+            <i-lucide-plus />
+          </el-icon>
+          {t('packages_form_aggregate_add_agg_field')}
+        </ElButton>
+      </div>
+    )
+  },
+})