wuweihua/BMS
2
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

拉取上游数据记录es里面

Browse Source
This commit is contained in:
wuweihua
2026-10-29 18:17:47 +08:00
parent 02c49e3efe
commit 6b10a41b9d
5 changed files with 136 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bms-back/bms-admin/src/main/java/com/bms/web/controller/monitor/TabOrderMonitorController.java
+6 -5
View File
@@ -318,13 +318,14 @@ public class TabOrderMonitorController extends BaseController
{
continue;
}
if (StringUtils.isNotEmpty(row.getOrderId()))
{
return prefix + "_" + row.getOrderId();
}
// Prefer transferNo as bizId (same as BILL / upstream sync ES).
if (StringUtils.isNotEmpty(row.getTransferOrderId()))
{
return prefix + "_" + row.getTransferOrderId();
return prefix + "_" + row.getTransferOrderId().trim();
}
if (StringUtils.isNotEmpty(row.getOrderId()))
{
return prefix + "_" + row.getOrderId().trim();
}
}
}
bms-back/bms-system/src/main/java/com/bms/price/search/EsBillSearchStore.java
+18 -1
View File
@@ -66,7 +66,8 @@ public class EsBillSearchStore implements IBillSearchStore
doc.put("status", StringUtils.isEmpty(status) ? EsBizLogSupport.STATUS_SUCCESS : status.trim());
doc.put("error", StringUtils.isEmpty(errorMessage) ? null : errorMessage);
doc.put("request", parseOrRaw(requestJson));
doc.put("response", parseOrRaw(responseJson));
// Normalize response.ok boolean -> 1/0: index mapping typed response.ok as long.
doc.put("response", normalizeResponseForEs(parseOrRaw(responseJson)));
String ts = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(now);
doc.put("updatedAt", ts);
doc.put("updateTime", ts);
@@ -81,6 +82,22 @@ public class EsBillSearchStore implements IBillSearchStore
}
}
/** Index historically mapped response.ok as long; coerce boolean to 1/0. */
private Object normalizeResponseForEs(Object response)
{
if (!(response instanceof JSONObject))
{
return response;
}
JSONObject obj = (JSONObject) response;
Object ok = obj.get("ok");
if (ok instanceof Boolean)
{
obj.put("ok", Boolean.TRUE.equals(ok) ? Integer.valueOf(1) : Integer.valueOf(0));
}
return obj;
}
@Override
public BillSyncSnapshot get(String logType, String bizId)
{
bms-back/bms-system/src/main/java/com/bms/price/search/TabOrderSyncEsLog.java
+21 -8
View File
@@ -46,27 +46,40 @@ public final class TabOrderSyncEsLog
/** Persist one failed sync order so it appears in order-source operation log. */
public static void saveOneFail(IBillSearchStore store, Object row, Object response, String errorMessage)
{
String transferNo = null;
String orderId = null;
if (row instanceof com.bms.price.domain.vo.TabOrderMonitorVo)
{
orderId = ((com.bms.price.domain.vo.TabOrderMonitorVo) row).getOrderId();
com.bms.price.domain.vo.TabOrderMonitorVo vo = (com.bms.price.domain.vo.TabOrderMonitorVo) row;
transferNo = vo.getTransferOrderId();
orderId = vo.getOrderId();
}
else if (row instanceof Map)
{
Object v = ((Map<?, ?>) row).get("orderId");
if (v != null)
Map<?, ?> map = (Map<?, ?>) row;
Object tf = map.get("transferOrderId");
if (tf == null)
{
orderId = String.valueOf(v);
tf = map.get("transferNo");
}
if (tf != null)
{
transferNo = String.valueOf(tf);
}
Object oid = map.get("orderId");
if (oid != null)
{
orderId = String.valueOf(oid);
}
}
String bizId;
if (StringUtils.isEmpty(orderId))
String bizId = EsBizLogSupport.billBizId(transferNo, orderId);
if (bizId != null && bizId.startsWith("FAIL-"))
{
bizId = "ONE_FAIL_" + System.currentTimeMillis();
}
else
else if (StringUtils.isNotEmpty(bizId))
{
bizId = "ONE_" + orderId.trim();
bizId = "ONE_" + bizId.trim();
}
save(store, bizId, "ONE_FAIL", row, response, EsBizLogSupport.STATUS_FAIL, errorMessage);
}
bms-back/bms-system/src/main/java/com/bms/price/service/impl/PrcBillingTaskServiceImpl.java
+60
View File
@@ -29,8 +29,12 @@ import com.bms.price.mapper.PrcBillingTaskMapper;
import com.bms.price.mapper.PrcOrderBillMapper;
import com.bms.price.mq.BillingTaskMessage;
import com.bms.price.mq.IBillingTaskMqPublisher;
import com.alibaba.fastjson2.JSONObject;
import com.bms.price.oracle.OracleOrderBillMapper;
import com.bms.price.oracle.OracleUpstreamOrderService;
import com.bms.price.search.EsBizLogSupport;
import com.bms.price.search.EsLogType;
import com.bms.price.search.IBillSearchStore;
import com.bms.price.service.IPrcBillingTaskService;
import com.bms.price.service.IPrcOrderBillService;
import com.bms.price.task.PrcBillingTaskRetryAsyncRunner;
@@ -64,6 +68,9 @@ public class PrcBillingTaskServiceImpl implements IPrcBillingTaskService
@Autowired
private PrcBillingTaskRetryAsyncRunner retryAsyncRunner;
@Autowired
private IBillSearchStore billSearchStore;
@Override
@Transactional(rollbackFor = Exception.class)
public PrcBillingTask submitInbound(BillingTaskSubmitRequest request)
@@ -265,6 +272,8 @@ public class PrcBillingTaskServiceImpl implements IPrcBillingTaskService
private void billFromUpstreamRow(PrcBillingTask task, Map<String, Object> row)
{
Long taskId = task.getTaskId();
// When caller already pulled the row (monitor sync), still write ES once here.
saveUpstreamPullEs(task, row);
TmsPriceBillRequest billReq = oracleOrderBillMapper.toBillRequest(row,
billingProperties.getOracle().getDefaultProductCode());
if (billReq == null)
@@ -660,6 +669,57 @@ public class PrcBillingTaskServiceImpl implements IPrcBillingTaskService
billingTaskMapper.updateBillingResult(upd);
}
/**
* Write pulled Oracle TAB_ORDER row to ES. bizId prefers transferNo.
* Always called once before bill/snapshot so pull is visible even when billing soft-fails.
*/
private void saveUpstreamPullEs(PrcBillingTask task, Map<String, Object> row)
{
if (billSearchStore == null || row == null || row.isEmpty())
{
return;
}
String transferNo = task == null ? null : trim(task.getTransferNo());
String orderNo = task == null ? null : trim(task.getOrderNo());
if (StringUtils.isEmpty(transferNo))
{
Object tf = row.get("TRANSFER_ORDER_ID");
if (tf != null)
{
transferNo = String.valueOf(tf).trim();
}
}
if (StringUtils.isEmpty(orderNo))
{
Object oid = row.get("ORDER_ID");
if (oid != null)
{
orderNo = String.valueOf(oid).trim();
}
}
String bizId = EsBizLogSupport.billBizId(transferNo, orderNo);
JSONObject req = new JSONObject();
req.put("action", "UPSTREAM_PULL");
if (task != null)
{
req.put("taskId", task.getTaskId());
req.put("billType", task.getBillType());
}
req.put("transferNo", transferNo);
req.put("orderNo", orderNo);
req.put("upstreamRow", row);
try
{
// ES mapping has response.ok as long; use 1/0 not boolean true/false.
billSearchStore.save(EsLogType.BILL, bizId, req.toJSONString(),
"{\"ok\":1,\"action\":\"UPSTREAM_PULL\"}");
}
catch (Exception e)
{
log.warn("saveUpstreamPullEs failed bizId={} msg={}", bizId, e.getMessage());
}
}
private static String trim(String s)
{
return s == null ? null : s.trim();
bms-back/bms-system/src/main/java/com/bms/price/service/impl/PrcOrderBillServiceImpl.java
+31 -2
View File
@@ -3043,8 +3043,12 @@ public class PrcOrderBillServiceImpl implements IPrcOrderBillService
req.put("billId", bill.getBillId());
req.put("orderStatus", bill.getOrderStatus());
req.put("creating", Boolean.valueOf(creating));
billSearchStore.save(EsLogType.BILL, bill.getBillId(), req.toJSONString(),
"{\"ok\":true,\"action\":\"UPSTREAM_SYNC\"}");
// Persist pulled Oracle TAB_ORDER row for ops; bizId = transferNo (fallback orderNo).
req.put("upstreamRow", upstreamRow);
String bizId = EsBizLogSupport.billBizId(bill.getTransferNo(), orderNo);
// ES mapping has response.ok as long; use 1/0 not boolean true/false.
billSearchStore.save(EsLogType.BILL, bizId, req.toJSONString(),
"{\"ok\":1,\"action\":\"UPSTREAM_SYNC\"}");
return bill.getBillId();
}
@@ -3171,6 +3175,19 @@ public class PrcOrderBillServiceImpl implements IPrcOrderBillService
incomeBillMapper.deleteByBillId(bill.getBillId());
}
persistBillExt(bill, request.getUpstreamRow());
// Soft-fail path also writes ES (bizId = transferNo) so upstream pull is searchable.
JSONObject failReq = new JSONObject();
failReq.put("action", "BILL_SOFT_FAIL");
failReq.put("orderNo", orderNo);
failReq.put("transferNo", request.getTransferNo());
failReq.put("billId", bill.getBillId());
failReq.put("billingFailStatus", Integer.valueOf(billingFailStatus));
failReq.put("errorMessage", errorMessage);
failReq.put("upstreamRow", request.getUpstreamRow());
billSearchStore.save(EsLogType.BILL, EsBizLogSupport.billBizId(request),
failReq.toJSONString(),
"{\"ok\":0,\"action\":\"BILL_SOFT_FAIL\",\"billingFailStatus\":" + billingFailStatus + "}",
EsBizLogSupport.STATUS_FAIL, errorMessage);
}
/**
@@ -3193,6 +3210,18 @@ public class PrcOrderBillServiceImpl implements IPrcOrderBillService
if (request != null)
{
persistBillExt(bill, request.getUpstreamRow());
JSONObject failReq = new JSONObject();
failReq.put("action", "BILL_SOFT_FAIL_LOCKED");
failReq.put("orderNo", bill.getOrderNo());
failReq.put("transferNo", request.getTransferNo());
failReq.put("billId", bill.getBillId());
failReq.put("billingFailStatus", Integer.valueOf(billingFailStatus));
failReq.put("errorMessage", errorMessage);
failReq.put("upstreamRow", request.getUpstreamRow());
billSearchStore.save(EsLogType.BILL, EsBizLogSupport.billBizId(request),
failReq.toJSONString(),
"{\"ok\":0,\"action\":\"BILL_SOFT_FAIL_LOCKED\",\"billingFailStatus\":" + billingFailStatus + "}",
EsBizLogSupport.STATUS_FAIL, errorMessage);
}
}