플로우 외부db연결

backend-node/src/services/flowDataMoveService.ts

@@ -6,17 +6,20 @@
  */
 
 import db from "../database/db";
-import { FlowAuditLog } from "../types/flow";
+import { FlowAuditLog, FlowIntegrationContext } from "../types/flow";
 import { FlowDefinitionService } from "./flowDefinitionService";
 import { FlowStepService } from "./flowStepService";
+import { FlowExternalDbIntegrationService } from "./flowExternalDbIntegrationService";
 
 export class FlowDataMoveService {
   private flowDefinitionService: FlowDefinitionService;
   private flowStepService: FlowStepService;
+  private externalDbIntegrationService: FlowExternalDbIntegrationService;
 
   constructor() {
     this.flowDefinitionService = new FlowDefinitionService();
     this.flowStepService = new FlowStepService();
+    this.externalDbIntegrationService = new FlowExternalDbIntegrationService();
   }
 
   /**
@@ -104,7 +107,23 @@ export class FlowDataMoveService {
           );
         }
 
-        // 4. 감사 로그 기록
+        // 4. 외부 DB 연동 실행 (설정된 경우)
+        if (
+          toStep.integrationType &&
+          toStep.integrationType !== "internal" &&
+          toStep.integrationConfig
+        ) {
+          await this.executeExternalIntegration(
+            toStep,
+            flowId,
+            targetDataId,
+            sourceTable,
+            userId,
+            additionalData
+          );
+        }
+
+        // 5. 감사 로그 기록
         await this.logDataMove(client, {
           flowId,
           fromStepId,
@@ -435,4 +454,140 @@ export class FlowDataMoveService {
       statusTo: row.status_to,
     }));
   }
+
+  /**
+   * 외부 DB 연동 실행
+   */
+  private async executeExternalIntegration(
+    toStep: any,
+    flowId: number,
+    dataId: any,
+    tableName: string | undefined,
+    userId: string,
+    additionalData?: Record<string, any>
+  ): Promise<void> {
+    const startTime = Date.now();
+
+    try {
+      // 연동 컨텍스트 구성
+      const context: FlowIntegrationContext = {
+        flowId,
+        stepId: toStep.id,
+        dataId,
+        tableName,
+        currentUser: userId,
+        variables: {
+          ...additionalData,
+          stepName: toStep.stepName,
+          stepId: toStep.id,
+        },
+      };
+
+      // 연동 타입별 처리
+      switch (toStep.integrationType) {
+        case "external_db":
+          const result = await this.externalDbIntegrationService.execute(
+            context,
+            toStep.integrationConfig
+          );
+
+          // 연동 로그 기록
+          await this.logIntegration(
+            flowId,
+            toStep.id,
+            dataId,
+            toStep.integrationType,
+            toStep.integrationConfig.connectionId,
+            toStep.integrationConfig,
+            result.data,
+            result.success ? "success" : "failed",
+            result.error?.message,
+            Date.now() - startTime,
+            userId
+          );
+
+          if (!result.success) {
+            throw new Error(
+              `외부 DB 연동 실패: ${result.error?.message || "알 수 없는 오류"}`
+            );
+          }
+          break;
+
+        case "rest_api":
+          // REST API 연동 (추후 구현)
+          console.warn("REST API 연동은 아직 구현되지 않았습니다");
+          break;
+
+        case "webhook":
+          // Webhook 연동 (추후 구현)
+          console.warn("Webhook 연동은 아직 구현되지 않았습니다");
+          break;
+
+        case "hybrid":
+          // 복합 연동 (추후 구현)
+          console.warn("복합 연동은 아직 구현되지 않았습니다");
+          break;
+
+        default:
+          throw new Error(`지원하지 않는 연동 타입: ${toStep.integrationType}`);
+      }
+    } catch (error: any) {
+      console.error("외부 연동 실행 실패:", error);
+      // 연동 실패 로그 기록
+      await this.logIntegration(
+        flowId,
+        toStep.id,
+        dataId,
+        toStep.integrationType,
+        toStep.integrationConfig?.connectionId,
+        toStep.integrationConfig,
+        null,
+        "failed",
+        error.message,
+        Date.now() - startTime,
+        userId
+      );
+      throw error;
+    }
+  }
+
+  /**
+   * 외부 연동 로그 기록
+   */
+  private async logIntegration(
+    flowId: number,
+    stepId: number,
+    dataId: any,
+    integrationType: string,
+    connectionId: number | undefined,
+    requestPayload: any,
+    responsePayload: any,
+    status: "success" | "failed" | "timeout" | "rollback",
+    errorMessage: string | undefined,
+    executionTimeMs: number,
+    userId: string
+  ): Promise<void> {
+    const query = `
+      INSERT INTO flow_integration_log (
+        flow_definition_id, step_id, data_id, integration_type, connection_id,
+        request_payload, response_payload, status, error_message, 
+        execution_time_ms, executed_by, executed_at
+      )
+      VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, NOW())
+    `;
+
+    await db.query(query, [
+      flowId,
+      stepId,
+      String(dataId),
+      integrationType,
+      connectionId || null,
+      requestPayload ? JSON.stringify(requestPayload) : null,
+      responsePayload ? JSON.stringify(responsePayload) : null,
+      status,
+      errorMessage || null,
+      executionTimeMs,
+      userId,
+    ]);
+  }
 }