chpark/vexplor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

플로우 외부연결 중간커밋

Browse Source
This commit is contained in:
kjs
2025-10-21 13:19:18 +09:00
parent 967f9a9f5b
commit 0d96ea566b
12 changed files with 1667 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

351
backend-node/src/services/flowDataMoveService.ts
View File

@@ -6,10 +6,25 @@
*/
import db from "../database/db";
import { FlowAuditLog, FlowIntegrationContext } from "../types/flow";
import {
FlowAuditLog,
FlowIntegrationContext,
FlowDefinition,
} from "../types/flow";
import { FlowDefinitionService } from "./flowDefinitionService";
import { FlowStepService } from "./flowStepService";
import { FlowExternalDbIntegrationService } from "./flowExternalDbIntegrationService";
import {
getExternalPool,
executeExternalQuery,
executeExternalTransaction,
} from "./externalDbHelper";
import {
getPlaceholder,
buildUpdateQuery,
buildInsertQuery,
buildSelectQuery,
} from "./dbQueryBuilder";
export class FlowDataMoveService {
private flowDefinitionService: FlowDefinitionService;
@@ -33,6 +48,28 @@ export class FlowDataMoveService {
userId: string = "system",
additionalData?: Record<string, any>
): Promise<{ success: boolean; targetDataId?: any; message?: string }> {
// 0. 플로우 정의 조회 (DB 소스 확인)
const flowDefinition = await this.flowDefinitionService.findById(flowId);
if (!flowDefinition) {
throw new Error(`플로우를 찾을 수 없습니다 (ID: ${flowId})`);
}
// 외부 DB인 경우 별도 처리
if (
flowDefinition.dbSourceType === "external" &&
flowDefinition.dbConnectionId
) {
return await this.moveDataToStepExternal(
flowDefinition.dbConnectionId,
fromStepId,
toStepId,
dataId,
userId,
additionalData
);
}
// 내부 DB 처리 (기존 로직)
return await db.transaction(async (client) => {
try {
// 1. 단계 정보 조회
@@ -160,7 +197,14 @@ export class FlowDataMoveService {
dataId: any,
additionalData?: Record<string, any>
): Promise<void> {
const statusColumn = toStep.statusColumn || "flow_status";
// 상태 컬럼이 지정되지 않은 경우 에러
if (!toStep.statusColumn) {
throw new Error(
`단계 "${toStep.stepName}"의 상태 컬럼이 지정되지 않았습니다. 플로우 편집 화면에서 "상태 컬럼명"을 설정해주세요.`
);
}
const statusColumn = toStep.statusColumn;
const tableName = fromStep.tableName;
// 추가 필드 업데이트 준비
@@ -590,4 +634,307 @@ export class FlowDataMoveService {
userId,
]);
}
/**
* 외부 DB 데이터 이동 처리
*/
private async moveDataToStepExternal(
dbConnectionId: number,
fromStepId: number,
toStepId: number,
dataId: any,
userId: string = "system",
additionalData?: Record<string, any>
): Promise<{ success: boolean; targetDataId?: any; message?: string }> {
return await executeExternalTransaction(
dbConnectionId,
async (externalClient, dbType) => {
try {
// 1. 단계 정보 조회 (내부 DB에서)
const fromStep = await this.flowStepService.findById(fromStepId);
const toStep = await this.flowStepService.findById(toStepId);
if (!fromStep || !toStep) {
throw new Error("유효하지 않은 단계입니다");
}
let targetDataId = dataId;
let sourceTable = fromStep.tableName;
let targetTable = toStep.tableName || fromStep.tableName;
// 2. 이동 방식에 따라 처리
switch (toStep.moveType || "status") {
case "status":
// 상태 변경 방식
await this.moveByStatusChangeExternal(
externalClient,
dbType,
fromStep,
toStep,
dataId,
additionalData
);
break;
case "table":
// 테이블 이동 방식
targetDataId = await this.moveByTableTransferExternal(
externalClient,
dbType,
fromStep,
toStep,
dataId,
additionalData
);
targetTable = toStep.targetTable || toStep.tableName;
break;
case "both":
// 하이브리드 방식: 둘 다 수행
await this.moveByStatusChangeExternal(
externalClient,
dbType,
fromStep,
toStep,
dataId,
additionalData
);
targetDataId = await this.moveByTableTransferExternal(
externalClient,
dbType,
fromStep,
toStep,
dataId,
additionalData
);
targetTable = toStep.targetTable || toStep.tableName;
break;
default:
throw new Error(
`지원하지 않는 이동 방식입니다: ${toStep.moveType}`
);
}
// 3. 외부 연동 처리는 생략 (외부 DB 자체가 외부이므로)
// 4. 감사 로그 기록 (내부 DB에)
// 외부 DB는 내부 DB 트랜잭션 외부이므로 직접 쿼리 실행
const auditQuery = `
INSERT INTO flow_audit_log (
flow_definition_id, from_step_id, to_step_id,
move_type, source_table, target_table,
source_data_id, target_data_id,
status_from, status_to,
changed_by, note
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
`;
await db.query(auditQuery, [
toStep.flowDefinitionId,
fromStep.id,
toStep.id,
toStep.moveType || "status",
sourceTable,
targetTable,
dataId,
targetDataId,
null, // statusFrom
toStep.statusValue || null, // statusTo
userId,
`외부 DB (${dbType}) 데이터 이동`,
]);
return {
success: true,
targetDataId,
message: `데이터 이동이 완료되었습니다 (외부 DB: ${dbType})`,
};
} catch (error: any) {
console.error("외부 DB 데이터 이동 오류:", error);
throw error;
}
}
);
}
/**
* 외부 DB 상태 변경 방식으로 데이터 이동
*/
private async moveByStatusChangeExternal(
externalClient: any,
dbType: string,
fromStep: any,
toStep: any,
dataId: any,
additionalData?: Record<string, any>
): Promise<void> {
// 상태 컬럼이 지정되지 않은 경우 에러
if (!toStep.statusColumn) {
throw new Error(
`단계 "${toStep.stepName}"의 상태 컬럼이 지정되지 않았습니다. 플로우 편집 화면에서 "상태 컬럼명"을 설정해주세요.`
);
}
const statusColumn = toStep.statusColumn;
const tableName = fromStep.tableName;
const normalizedDbType = dbType.toLowerCase();
// 업데이트할 필드 준비
const updateFields: { column: string; value: any }[] = [
{ column: statusColumn, value: toStep.statusValue },
];
// 추가 데이터가 있으면 함께 업데이트
if (additionalData) {
for (const [key, value] of Object.entries(additionalData)) {
updateFields.push({ column: key, value });
}
}
// DB별 쿼리 생성
const { query: updateQuery, values } = buildUpdateQuery(
dbType,
tableName,
updateFields,
"id"
);
// WHERE 절 값 설정 (마지막 파라미터)
values[values.length - 1] = dataId;
// 쿼리 실행 (DB 타입별 처리)
let result: any;
if (normalizedDbType === "postgresql") {
result = await externalClient.query(updateQuery, values);
} else if (normalizedDbType === "mysql" || normalizedDbType === "mariadb") {
[result] = await externalClient.query(updateQuery, values);
} else if (normalizedDbType === "mssql") {
const request = externalClient.request();
values.forEach((val: any, idx: number) => {
request.input(`p${idx + 1}`, val);
});
result = await request.query(updateQuery);
} else if (normalizedDbType === "oracle") {
result = await externalClient.execute(updateQuery, values, {
autoCommit: false,
});
}
// 결과 확인
const affectedRows =
normalizedDbType === "postgresql"
? result.rowCount
: normalizedDbType === "mssql"
? result.rowsAffected[0]
: normalizedDbType === "oracle"
? result.rowsAffected
: result.affectedRows;
if (affectedRows === 0) {
throw new Error(`데이터를 찾을 수 없습니다: ${dataId}`);
}
}
/**
* 외부 DB 테이블 이동 방식으로 데이터 이동
*/
private async moveByTableTransferExternal(
externalClient: any,
dbType: string,
fromStep: any,
toStep: any,
dataId: any,
additionalData?: Record<string, any>
): Promise<any> {
const sourceTable = fromStep.tableName;
const targetTable = toStep.targetTable || toStep.tableName;
const fieldMappings = toStep.fieldMappings || {};
const normalizedDbType = dbType.toLowerCase();
// 1. 소스 데이터 조회
const { query: selectQuery, placeholder } = buildSelectQuery(
dbType,
sourceTable,
"id"
);
let sourceResult: any;
if (normalizedDbType === "postgresql") {
sourceResult = await externalClient.query(selectQuery, [dataId]);
} else if (normalizedDbType === "mysql" || normalizedDbType === "mariadb") {
[sourceResult] = await externalClient.query(selectQuery, [dataId]);
} else if (normalizedDbType === "mssql") {
const request = externalClient.request();
request.input("p1", dataId);
sourceResult = await request.query(selectQuery);
sourceResult = { rows: sourceResult.recordset };
} else if (normalizedDbType === "oracle") {
sourceResult = await externalClient.execute(selectQuery, [dataId], {
autoCommit: false,
outFormat: 4001, // oracledb.OUT_FORMAT_OBJECT
});
}
const rows = sourceResult.rows || sourceResult;
if (!rows || rows.length === 0) {
throw new Error(`소스 데이터를 찾을 수 없습니다: ${dataId}`);
}
const sourceData = rows[0];
// 2. 필드 매핑 적용
const targetData: Record<string, any> = {};
for (const [targetField, sourceField] of Object.entries(fieldMappings)) {
const sourceFieldKey = sourceField as string;
if (sourceData[sourceFieldKey] !== undefined) {
targetData[targetField] = sourceData[sourceFieldKey];
}
}
// 추가 데이터 병합
if (additionalData) {
Object.assign(targetData, additionalData);
}
// 3. 대상 테이블에 삽입
const { query: insertQuery, values } = buildInsertQuery(
dbType,
targetTable,
targetData
);
let insertResult: any;
let newDataId: any;
if (normalizedDbType === "postgresql") {
insertResult = await externalClient.query(insertQuery, values);
newDataId = insertResult.rows[0].id;
} else if (normalizedDbType === "mysql" || normalizedDbType === "mariadb") {
[insertResult] = await externalClient.query(insertQuery, values);
newDataId = insertResult.insertId;
} else if (normalizedDbType === "mssql") {
const request = externalClient.request();
values.forEach((val: any, idx: number) => {
request.input(`p${idx + 1}`, val);
});
insertResult = await request.query(insertQuery);
newDataId = insertResult.recordset[0].id;
} else if (normalizedDbType === "oracle") {
// Oracle RETURNING 절 처리
const outBinds: any = { id: { dir: 3003, type: 2001 } }; // OUT, NUMBER
insertResult = await externalClient.execute(insertQuery, values, {
autoCommit: false,
outBinds: outBinds,
});
newDataId = insertResult.outBinds.id[0];
}
// 4. 필요 시 소스 데이터 삭제 (옵션)
// const deletePlaceholder = getPlaceholder(dbType, 1);
// await externalClient.query(`DELETE FROM ${sourceTable} WHERE id = ${deletePlaceholder}`, [dataId]);
return newDataId;
}
}