refactor(core):重构消息处理流程并优化数据映射逻辑

- 移除了 BaseCustomer 中的 currency 字段及相关注释 - 在 CustomerBuilder 中启用 systemDataId 非空校验 - 删除 CustomerProxy 和 CustomerSender 中的 thirdPartyCode 字段 - 调整 DynamicTaxPipelineMappingService 的包引用路径并移除默认状态设置 - 简化 InitializeProcessor 的依赖引入和处理逻辑- 在 MessageContext 中新增 pipelineDataId、systemDataId、tenantPipeline 等字段及枚举类型 - 优化 ProxyProcessor 处理流程，减少冗余代码 - 扩展 TaxPipelineMapping 及其创建对象，增加 errorMessage 字段支持错误记录 - 更新数据库表结构，添加 error_message 字段并调整状态值含义 - 新增 TaxReceiveService 统一处理消息接收、转换与持久化操作 - 增加新的税务系统异常类型 PARAMETER_IS_NOT_PARSED_CORRECTLY

refactor(core):重构消息处理流程并优化数据映射逻辑
- 移除了 BaseCustomer 中的 currency 字段及相关注释 - 在 CustomerBuilder 中启用 systemDataId 非空校验 - 删除 CustomerProxy 和 CustomerSender 中的 thirdPartyCode 字段 - 调整 DynamicTaxPipelineMappingService 的包引用路径并移除默认状态设置 - 简化 InitializeProcessor 的依赖引入和处理逻辑- 在 MessageContext 中新增 pipelineDataId、systemDataId、tenantPipeline 等字段及枚举类型 - 优化 ProxyProcessor 处理流程，减少冗余代码 - 扩展 TaxPipelineMapping 及其创建对象，增加 errorMessage 字段支持错误记录 - 更新数据库表结构，添加 error_message 字段并调整状态值含义 - 新增 TaxReceiveService 统一处理消息接收、转换与持久化操作 - 增加新的税务系统异常类型 PARAMETER_IS_NOT_PARSED_CORRECTLY
zhangmeiyang
1 parent 57a36eb8
Showing 18 changed files with 158 additions and 98 deletions
tax-boot/src/main/java/com/diligrp/tax/boot/receiver/TaxReceiver.java
tax-boot/src/main/java/com/diligrp/tax/boot/service/TaxReceiveService.java
tax-central/src/main/java/com/diligrp/tax/central/domain/MessageContext.java
tax-central/src/main/java/com/diligrp/tax/central/domain/document/kingdee/BaseCustomer.java
tax-central/src/main/java/com/diligrp/tax/central/domain/proxy/kingdee/CustomerProxy.java
tax-central/src/main/java/com/diligrp/tax/central/model/TaxPipelineMappingCreate.java
tax-central/src/main/java/com/diligrp/tax/central/model/TenantTaxPipelineMapping.java
tax-central/src/main/java/com/diligrp/tax/central/service/ITaxPipelineMappingService.java
tax-storage/src/main/java/com/diligrp/tax/storage/type/MappingStateType.java → tax-central/src/main/java/com/diligrp/tax/central/type/MappingStateType.java
tax-central/src/main/java/com/diligrp/tax/central/type/TaxSystemType.java
tax-doc/src/main/java/com/diligrp/tax/doc/demarcate/kingdee/CustomerBuilder.java
tax-doc/src/main/java/com/diligrp/tax/doc/process/kingdee/InitializeProcessor.java
tax-map/src/main/java/com/diligrp/tax/mapping/process/kingdee/MappingProcessor.java
tax-proxy/src/main/java/com/diligrp/tax/proxy/demarcate/kingdee/CustomerSender.java
tax-proxy/src/main/java/com/diligrp/tax/proxy/process/kingdee/ProxyProcessor.java
tax-storage/src/main/java/com/diligrp/tax/storage/domain/TaxPipelineMapping.java
tax-storage/src/main/java/com/diligrp/tax/storage/service/DynamicTaxPipelineMappingService.java
tax-storage/src/main/resources/com/diligrp/tax/storage/repo/TaxPipelineMappingRepository.xml
 package com.diligrp.tax.boot.receiver;
+import com.diligrp.tax.boot.service.TaxReceiveService;
 import com.diligrp.tax.central.domain.MessageContext;
-import com.diligrp.tax.central.process.ProcessorChain;
-import com.diligrp.tax.central.type.StatusType;
-import com.diligrp.tax.central.type.SystemType;
-import com.diligrp.tax.central.utils.JsonUtils;
-import com.diligrp.tax.storage.service.TenantTaxService;
 import com.rabbitmq.client.Channel;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
@@ -15,12 +11,11 @@ import org.springframework.amqp.rabbit.annotation.Queue;
 import org.springframework.amqp.rabbit.annotation.QueueBinding;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
 import java.io.IOException;
-import java.io.PrintWriter;
-import java.io.StringWriter;
 import java.nio.charset.StandardCharsets;
-import java.util.Map;
+import java.util.Optional;
 import static com.diligrp.tax.boot.queue.TaxAutoPush.*;
@@ -35,10 +30,7 @@ import static com.diligrp.tax.boot.queue.TaxAutoPush.*;
 public class TaxReceiver {
     @Resource
-    private Map<SystemType, ProcessorChain> processorChainMap;
-
-    @Resource
-    private TenantTaxService tenantTaxService;
+    private TaxReceiveService taxReceiveService;
     @RabbitListener(bindings =
     @QueueBinding(
@@ -47,34 +39,20 @@ public class TaxReceiver {
             key = NORMAL_ROUTING),
             ackMode = "MANUAL"
     )
+    @Transactional
     public void receiveMessage(Channel channel, Message message) throws IOException {
         var content = new String(message.getBody(), StandardCharsets.UTF_8);
         log.info("tax-agent收到消息:{}", content);
-        MessageContext ctx = JsonUtils.fromJsonString(content, MessageContext.class);
+        MessageContext ctx = null;
         try {
-            handle(ctx);
+            ctx = taxReceiveService.messageHandle(content);
+            taxReceiveService.recordMapping(ctx);
             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
             log.info("tax-agent消息处理成功:{}", content);
         } catch (Exception e) {
             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
-            recordError(e, ctx);
+            Optional.ofNullable(ctx).ifPresentOrElse(c -> taxReceiveService.recordMappingError(e, c), () -> log.error("tax-agent解析错误:", e));
             log.error("tax-agent消息处理失败:", e);
         }
     }
-
-    private void handle(MessageContext ctx) {
-        ctx.setTenantId(tenantTaxService.getTenantId(ctx.getGroup(), ctx.getEntity()));
-        MessageContext messageContext = processorChainMap.get(SystemType.from(ctx.getSystemType())).startProcess(ctx);
-        messageContext.setStatus(StatusType.SUCCESS.code);
-        //TODO write to db
-    }
-
-    private void recordError(Exception e, MessageContext ctx) {
-        StringWriter sw = new StringWriter();
-        PrintWriter pw = new PrintWriter(sw);
-        e.printStackTrace(pw);
-        ctx.setError(sw.toString());
-        ctx.setStatus(StatusType.FAIL.code);
-        //TODO write error to db
-    }
 }
+package com.diligrp.tax.boot.service;
+
+import com.diligrp.tax.central.context.TenantStorageContext;
+import com.diligrp.tax.central.domain.MessageContext;
+import com.diligrp.tax.central.exception.TaxAgentServiceException;
+import com.diligrp.tax.central.model.TaxPipelineMappingCreate;
+import com.diligrp.tax.central.model.TenantPipeline;
+import com.diligrp.tax.central.process.ProcessorChain;
+import com.diligrp.tax.central.service.ITaxPipelineMappingService;
+import com.diligrp.tax.central.service.ITenantTaxService;
+import com.diligrp.tax.central.type.DocumentType;
+import com.diligrp.tax.central.type.MappingStateType;
+import com.diligrp.tax.central.type.SystemType;
+import com.diligrp.tax.central.type.TaxSystemType;
+import com.diligrp.tax.central.utils.JsonUtils;
+import jakarta.annotation.Resource;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * @Author: zhangmeiyang
+ * @CreateTime: 2025-11-07 17:22
+ * @Version: todo
+ */
+@Service
+public class TaxReceiveService {
+
+    @Resource
+    private Map<SystemType, ProcessorChain> processorChainMap;
+
+    @Resource
+    private ITenantTaxService tenantTaxService;
+
+    @Resource
+    private ITaxPipelineMappingService taxPipelineMappingService;
+
+
+    public MessageContext messageHandle(String content) {
+        MessageContext ctx = JsonUtils.fromJsonString(content, MessageContext.class);
+        DocumentType from = DocumentType.from(ctx.getDocumentType());
+        SystemType system = SystemType.from(ctx.getSystemType());
+        Optional.of(system).filter(s -> s.documentTypes.contains(from)).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.BUSINESS_MATCHES_ARE_INCORRECT));
+        // 校验业务类型和系统类型
+        ctx.setDocumentTypeEnum(from);
+        ctx.setSystemTypeEnum(system);
+        Optional<TenantPipeline> option = TenantStorageContext.getTenantPipeline(ctx.getTenantId(), system, ctx.getPipelineCode());
+        TenantPipeline pipeline = option.orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.NO_MATCHING_SET_OF_ACCOUNTS_FOUND));
+        //获取租户账套
+        ctx.setTenantPipeline(pipeline);
+        //获取租户id
+        ctx.setTenantId(tenantTaxService.getTenantId(ctx.getGroup(), ctx.getEntity()));
+        return processorChainMap.get(SystemType.from(ctx.getSystemType())).startProcess(ctx);
+    }
+
+    @Transactional
+    public void recordMapping(MessageContext messageContext) {
+        TaxPipelineMappingCreate create = new TaxPipelineMappingCreate();
+        create.setTenantId(messageContext.getTenantId());
+        create.setPipelineId(messageContext.getTenantPipeline().getId());
+        create.setDocumentType(messageContext.getDocumentType());
+        create.setSystemDataId(messageContext.getSystemDataId());
+        create.setState(MappingStateType.SYNCED.value);
+        create.setOriginData(messageContext.getMsgBody());
+        taxPipelineMappingService.insert(create);
+    }
+
+    @Transactional
+    public void recordMappingError(Exception e, MessageContext ctx) {
+        StringWriter sw = new StringWriter();
+        PrintWriter pw = new PrintWriter(sw);
+        e.printStackTrace(pw);
+        TaxPipelineMappingCreate create = new TaxPipelineMappingCreate();
+        create.setTenantId(ctx.getTenantId());
+        create.setPipelineId(ctx.getTenantPipeline().getId());
+        create.setDocumentType(ctx.getDocumentType());
+        create.setSystemDataId(ctx.getSystemDataId());
+        create.setPipelineDataId(ctx.getPipelineDataId());
+        create.setOriginData(ctx.getMsgBody());
+        create.setState(MappingStateType.SYNC_FAILED.value);
+        create.setErrorMessage(sw.toString());
+        taxPipelineMappingService.insert(create);
+    }
+
+}
 package com.diligrp.tax.central.domain;
+import com.diligrp.tax.central.model.TenantPipeline;
+import com.diligrp.tax.central.type.DocumentType;
+import com.diligrp.tax.central.type.SystemType;
 import lombok.Getter;
 import lombok.Setter;
@@ -7,9 +10,18 @@ import lombok.Setter;
 @Getter
 public class MessageContext {
     /**
-     * 开放 ID
+     * 账套数据写入id
      */
-    private String thirdPartyCode;
+    private String pipelineDataId;
+
+    /**
+     * 账套数据对接第三方系统id
+     */
+    private String systemDataId;
+    /**
+     * 租户管道
+     */
+    private TenantPipeline tenantPipeline;
     /**
      * 分组
      */
@@ -51,11 +63,12 @@ public class MessageContext {
      */
     private String documentType;
     /**
-     * 错误
+     * 文档类型枚举
      */
-    private String error;
+    private DocumentType documentTypeEnum;
+
     /**
-     * 成功标志
+     * 系统类型枚举
      */
-    private Integer status;
+    private SystemType systemTypeEnum;
 }
@@ -18,11 +18,6 @@ public abstract class BaseCustomer extends BaseDocument {
      * 系统数据 ID
      */
     private String systemDataId;
-//    /**
-//     * 结算货币(FTRADINGCURRID)
-//     */
-//    @Converter(value = CustomerCurrencyConverter.class, targetField = "FTRADINGCURRID")
-//    protected String currency;
     /**
      * 客户 ID(FCUSTID)
      */
@@ -13,5 +13,4 @@ import lombok.Setter;
 @Setter
 public class CustomerProxy extends BaseProxy {
     private String thirdPartyId;
-    private String thirdPartyCode;
 }
@@ -50,4 +50,9 @@ public class TaxPipelineMappingCreate {
      */
     @NotNull(message = "同步状态不能为空")
     private Integer state;
+
+    /**
+     * 错误信息
+     */
+    private String errorMessage;
 }
@@ -33,6 +33,10 @@ public class TenantTaxPipelineMapping {
      */
     private String originData;
     /**
+     * 错误信息
+     */
+    private String errorMessage;
+    /**
      * 州
      */
     private Integer state;
 package com.diligrp.tax.central.service;
+import com.diligrp.tax.central.domain.MessageContext;
 import com.diligrp.tax.central.model.TaxPipelineMappingCreate;
 import com.diligrp.tax.central.model.TenantTaxPipelineMapping;
-package com.diligrp.tax.storage.type;
+package com.diligrp.tax.central.type;
 public enum MappingStateType {
     SYNCED(0, "已同步"),
@@ -18,6 +18,7 @@ public enum TaxSystemType {
     INVALID_HTTP_REQUEST_PARAMS(5007, "无效的http请求参数"),
     NO_MATCHING_SET_OF_ACCOUNTS_FOUND(5008, "未找到匹配账套"),
     NO_TENANT_INFORMATION_FOUND(5009, "未找到匹配账套"),
+    PARAMETER_IS_NOT_PARSED_CORRECTLY(5010, "参数解析不正确"),
     ;
     public final int code;
     public final String message;
@@ -38,7 +38,7 @@ public class CustomerBuilder extends Builder&lt;StandardCustomer&gt; {
     public StandardCustomer build(String body, TenantPipeline pipeline) {
         StandardCustomer customer = JsonUtils.fromJsonString(body, StandardCustomer.class);
         Optional.ofNullable(customer.getContacts()).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "请完善联系人信息"));
-//        Optional.ofNullable(customer.getSystemDataId()).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "请填写系统数据ID"));
+        Optional.ofNullable(customer.getSystemDataId()).orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.MISSING_BUSINESS_INFORMATION, "请填写系统数据ID"));
         List<TenantPipelineConfig> list = pipeline.getTenantPipelineConfigs().get(markDocument());
         Optional.ofNullable(list).ifPresent(ts -> configureDefaultData(customer, ts));
         //TODO 查询数据库的客户信息 如果存在 赋值id
 package com.diligrp.tax.doc.process.kingdee;
-import com.diligrp.tax.central.context.TenantStorageContext;
 import com.diligrp.tax.central.domain.MessageContext;
-import com.diligrp.tax.central.exception.TaxAgentServiceException;
-import com.diligrp.tax.central.model.TenantPipeline;
 import com.diligrp.tax.central.process.AbstractProcessor;
-import com.diligrp.tax.central.type.DocumentType;
-import com.diligrp.tax.central.type.SystemType;
-import com.diligrp.tax.central.type.TaxSystemType;
 import com.diligrp.tax.doc.context.DocumentContext;
 import com.diligrp.tax.doc.demarcate.Builder;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.core.annotation.Order;
 import org.springframework.stereotype.Component;
-import java.util.Optional;
-
 /**
  * 初始化处理器
@@ -30,15 +22,8 @@ public class InitializeProcessor extends AbstractProcessor {
     @Override
     public MessageContext process(MessageContext messageContext) {
-        DocumentType from = DocumentType.from(messageContext.getDocumentType());
-        SystemType system = SystemType.from(messageContext.getSystemType());
-        if (!system.documentTypes.contains(from)) {
-            throw new TaxAgentServiceException(TaxSystemType.BUSINESS_MATCHES_ARE_INCORRECT);
-        }
-        Optional<TenantPipeline> option = TenantStorageContext.getTenantPipeline(messageContext.getTenantId(), system, messageContext.getPipelineCode());
-        TenantPipeline pipeline = option.orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.NO_MATCHING_SET_OF_ACCOUNTS_FOUND));
-        Builder<?> builder = DocumentContext.CONTEXT.get(from);
-        messageContext.setDocumentObject(builder.build(messageContext.getMsgBody(),pipeline));
+        Builder<?> builder = DocumentContext.CONTEXT.get(messageContext.getDocumentTypeEnum());
+        messageContext.setDocumentObject(builder.build(messageContext.getMsgBody(), messageContext.getTenantPipeline()));
         return messageContext;
     }
 }
@@ -28,12 +28,7 @@ public class MappingProcessor extends AbstractProcessor {
     @Override
     public MessageContext process(MessageContext messageContext) {
-        DocumentType from = DocumentType.from(messageContext.getDocumentType());
-        SystemType system = SystemType.from(messageContext.getSystemType());
-        if (!system.documentTypes.contains(from)) {
-            throw new TaxAgentServiceException(TaxSystemType.BUSINESS_MATCHES_ARE_INCORRECT);
-        }
-        Transformer<?> transformer = MappingContext.CONTEXT.get(from);
+        Transformer<?> transformer = MappingContext.CONTEXT.get(messageContext.getDocumentTypeEnum());
         BaseMapping transform = transformer.transform(messageContext.getDocumentObject());
         messageContext.setMappingObject(transform);
         Map<String,Object> map = JsonUtils.convertValue(transform, new TypeReference<>() {});
@@ -50,7 +50,6 @@ public class CustomerSender extends Sender&lt;CustomerProxy&gt; {
         SuccessEntity first = successEntity.getFirst();
         var proxy = new CustomerProxy();
         proxy.setThirdPartyId(first.getId());
-        proxy.setThirdPartyCode(first.getNumber());
         return proxy;
     }
 package com.diligrp.tax.proxy.process.kingdee;
 import com.diligrp.tax.central.context.ConnectionContext;
-import com.diligrp.tax.central.context.TenantStorageContext;
 import com.diligrp.tax.central.domain.BaseProxy;
 import com.diligrp.tax.central.domain.MessageContext;
-import com.diligrp.tax.central.exception.TaxAgentServiceException;
 import com.diligrp.tax.central.manager.AbstractConnectionManager;
-import com.diligrp.tax.central.model.TenantPipeline;
 import com.diligrp.tax.central.process.AbstractProcessor;
-import com.diligrp.tax.central.type.DocumentType;
-import com.diligrp.tax.central.type.SystemType;
-import com.diligrp.tax.central.type.TaxSystemType;
 import com.diligrp.tax.proxy.context.ProxyContext;
 import com.kingdee.bos.webapi.entity.IdentifyInfo;
-import com.kingdee.bos.webapi.sdk.K3CloudApi;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.core.annotation.Order;
 import org.springframework.stereotype.Component;
-import java.util.Optional;
-
 /**
  * 代理发送处理器
  * 将Map<String, Object>发送到目标系统
@@ -31,13 +22,9 @@ public class ProxyProcessor extends AbstractProcessor {
     @Override
     public MessageContext process(MessageContext messageContext) {
-        DocumentType from = DocumentType.from(messageContext.getDocumentType());
-        SystemType systemType = SystemType.from(messageContext.getSystemType());
-        Optional<TenantPipeline> option = TenantStorageContext.getTenantPipeline(messageContext.getTenantId(), systemType, messageContext.getPipelineCode());
-        TenantPipeline pipeline = option.orElseThrow(() -> new TaxAgentServiceException(TaxSystemType.NO_MATCHING_SET_OF_ACCOUNTS_FOUND));
-        AbstractConnectionManager<?> abstractConnectionManager = ConnectionContext.CONNECTION_MAP.get(systemType);
-        IdentifyInfo identifyInfo = (IdentifyInfo) abstractConnectionManager.getConnection(pipeline);
-        BaseProxy send = ProxyContext.CONTEXT.get(from).send(messageContext.getMappingObject(), identifyInfo);
+        AbstractConnectionManager<?> abstractConnectionManager = ConnectionContext.CONNECTION_MAP.get(messageContext.getSystemTypeEnum());
+        IdentifyInfo identifyInfo = (IdentifyInfo) abstractConnectionManager.getConnection(messageContext.getTenantPipeline());
+        BaseProxy send = ProxyContext.CONTEXT.get(messageContext.getDocumentTypeEnum()).send(messageContext.getMappingObject(), identifyInfo);
         messageContext.setProxyObject(send);
         return messageContext;
     }
@@ -50,6 +50,11 @@ public class TaxPipelineMapping {
     private Integer state;
     /**
+     * 错误信息
+     */
+    private String errorMessage;
+
+    /**
      * 创建时间
      */
     private LocalDateTime createdTime;
@@ -11,7 +11,7 @@ import com.diligrp.tax.central.utils.JsonUtils;
 import com.diligrp.tax.storage.domain.TaxPipelineMapping;
 import com.diligrp.tax.storage.model.co.TaxPipelineMappingCO;
 import com.diligrp.tax.storage.repo.TaxPipelineMappingRepository;
-import com.diligrp.tax.storage.type.MappingStateType;
+import com.diligrp.tax.central.type.MappingStateType;
 import jakarta.annotation.Resource;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
@@ -86,7 +86,6 @@ public class DynamicTaxPipelineMappingService implements ITaxPipelineMappingServ
     public void insert(@Validated TaxPipelineMappingCreate taxPipelineMappingCreate) {
         DocumentType.validateDocumentType(taxPipelineMappingCreate.getDocumentType());
         TaxPipelineMapping taxPipelineMapping = JsonUtils.convertValue(taxPipelineMappingCreate, TaxPipelineMapping.class);
-        taxPipelineMapping.setState(MappingStateType.SYNCED.value);
         taxPipelineMappingRepository.insert(taxPipelineMapping);
     }
 }
@@ -8,14 +8,16 @@
                                             , system_data_id
                                             , pipeline_data_id
                                             , origin_data
-                                            , state)
+                                            , state
+                                            , error_message)
         VALUES
             ( #{pipelineId}
             , #{documentType}
             , #{systemDataId}
             , #{pipelineDataId}
             , #{originData}
-            , #{state})
+            , #{state}
+            , #{errorMessage})
     </insert>
     <update id="update">
         UPDATE
@@ -34,6 +36,7 @@
         , pipeline_data_id
         , origin_data
         , state
+        , error_message
         , created_Time
         , modified_Time
         FROM
@@ -55,6 +58,7 @@
           , pipeline_data_id
           , origin_data
           , state
+          , error_message
           , created_Time
           , modified_Time
         FROM
@@ -70,15 +74,16 @@
     <update id="createTenantMappingTable">
         CREATE TABLE ${tableName}
         (
-            `id`               bigint unsigned                                              NOT NULL AUTO_INCREMENT,
-            `pipeline_id`      bigint unsigned                                              NOT NULL COMMENT '账套ID',
-            `document_type`    varchar(40)                                                  NOT NULL COMMENT '业务单据类型',
-            `system_data_id`   varchar(50)                                                  NOT NULL COMMENT '系统数据ID',
-            `pipeline_data_id` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT '账套数据ID',
-            `origin_data`      json                                                         NOT NULL COMMENT '原始数据',
-            `state`            tinyint                                                      NOT NULL COMMENT '状态 1=同步成功，2=同步失败',
-            `created_time`     datetime                                                     NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
-            `modified_time`    datetime                                                     NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+            `id`               bigint unsigned NOT NULL AUTO_INCREMENT,
+            `pipeline_id`      bigint unsigned NOT NULL COMMENT '账套ID',
+            `document_type`    varchar(40)     NOT NULL COMMENT '业务单据类型',
+            `system_data_id`   varchar(50)     NOT NULL COMMENT '系统数据ID',
+            `pipeline_data_id` varchar(50)     NOT NULL COMMENT '账套数据ID',
+            `origin_data`      json            NOT NULL COMMENT '原始数据',
+            `error_message`    text            NULL COMMENT '错误信息',
+            `state`            tinyint         NOT NULL COMMENT '状态 0=已同步，1=同步失败，2=同步重试',
+            `created_time`     datetime        NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+            `modified_time`    datetime        NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
             PRIMARY KEY (`id`) USING BTREE,
             UNIQUE KEY `uk_pipeline_id_document_type_system_data_id` (`pipeline_id`, `document_type`, `system_data_id`) USING BTREE,
             KEY `idx_pipeline_id_state` (`pipeline_id`, `state`) USING BTREE