419 lines
18 KiB
C#
419 lines
18 KiB
C#
using System.Runtime.ExceptionServices;
|
||
using Fake.DependencyInjection;
|
||
using InterfaceForward.Application.Contracts;
|
||
using InterfaceForward.Application.Contracts.ForwardCore;
|
||
using InterfaceForward.Application.ForwardCore;
|
||
using InterfaceForward.Application.Helpers;
|
||
using InterfaceForward.Domain.Shared;
|
||
using InterfaceForward.Domain.Shared.Dtos;
|
||
using InterfaceForward.Domain.Shared.Enum;
|
||
using InterfaceForward.Repositories;
|
||
using InterfaceForward.Repositories.Log.Services;
|
||
using Microsoft.Extensions.DependencyInjection;
|
||
using Microsoft.Extensions.Options;
|
||
using Newtonsoft.Json.Linq;
|
||
using SJZY.InterfaceRelay.Repository.Log.ValueObjects;
|
||
|
||
namespace InterfaceForward.Application.Services;
|
||
|
||
public class InterfaceForwardCommon : ITransientDependency
|
||
{
|
||
/// <summary>
|
||
/// 上下文缓存数据：映射，账户
|
||
/// </summary>
|
||
public static readonly int ContextCacheTtl = int.MaxValue;
|
||
|
||
private readonly Dictionary<string, IForwardFlow> _flows = new();
|
||
|
||
private readonly InterfaceForwardQuery _interfaceForwardQuery;
|
||
private readonly InterfaceRelayOptions _interfaceRelayOptions;
|
||
private readonly ILogRepository _logRepository;
|
||
// private readonly IEntrySegmentContextAccessor _segContext;
|
||
private readonly IServiceProvider _serviceProvider;
|
||
|
||
public InterfaceForwardCommon(InterfaceForwardQuery interfaceForwardQuery
|
||
, ILogRepository logRepository
|
||
// , IEntrySegmentContextAccessor segContext
|
||
, IServiceProvider serviceProvider
|
||
, IOptions<InterfaceRelayOptions> options)
|
||
{
|
||
_interfaceForwardQuery = interfaceForwardQuery;
|
||
_logRepository = logRepository;
|
||
// _segContext = segContext;
|
||
_serviceProvider = serviceProvider;
|
||
_interfaceRelayOptions = options.Value;
|
||
}
|
||
|
||
public async Task<List<ForwardCoreContext>> VerifyBusinessAndBuildForwardContexts(string upStreamCode,
|
||
string serviceProviderCode)
|
||
{
|
||
// tips：上下文的缓存不维护，在发布时统一替代。
|
||
var contextCacheKey = GetContextCacheKey(upStreamCode, serviceProviderCode);
|
||
// _segContext.Context.Span.AddLog(LogEvent.Message("【转发核心上下文】开始构建"));
|
||
var cacheContext = (await RedisHelper.Client.GetAsync<string>(contextCacheKey))
|
||
.ToObject<ForwardCoreContextCache>();
|
||
if (cacheContext == null)
|
||
{
|
||
// _segContext.Context.Span.AddLog(LogEvent.Message("【转发核心上下文】没能命中缓存，重新构建"));
|
||
|
||
var systemInterface = await _interfaceForwardQuery.GetInterfaceByCodeAsync(upStreamCode);
|
||
if (systemInterface == null) throw new BusinessException(message: "系统接口不存在");
|
||
|
||
_logRepository.AppRequestLog.InterfaceCode = systemInterface.Code;
|
||
_logRepository.AppRequestLog.InterfaceName = systemInterface.Name;
|
||
|
||
var serviceProvider = await _interfaceForwardQuery.GetServiceProviderByCodeAsync(serviceProviderCode);
|
||
if (serviceProvider == null) throw new BusinessException(message: "服务商不存在");
|
||
cacheContext = await _interfaceForwardQuery.BuildForwardCoreContext(systemInterface, serviceProvider);
|
||
await RedisHelper.Client.SetAsync(contextCacheKey, cacheContext.ToJson(), ContextCacheTtl);
|
||
}
|
||
|
||
// _segContext.Context.Span.AddLog(LogEvent.Message("【转发核心上下文】构建完成"));
|
||
// 记录系统接口日志
|
||
_logRepository.AppRequestLog.InterfaceCode = cacheContext.OriginalInterface.Code;
|
||
_logRepository.AppRequestLog.InterfaceName = cacheContext.OriginalInterface.Name;
|
||
|
||
// tips：服务商主体账户的缓存实时维护
|
||
// _segContext.Context.Span.AddLog(LogEvent.Message("【服务商应用账户】开始构建"));
|
||
var accountCacheKey = GetAccountCacheKey(serviceProviderCode, _logRepository.AppId);
|
||
var accountFieldCache = (await RedisHelper.Client.GetAsync<string>(accountCacheKey))
|
||
.ToObject<List<ServiceProviderAccountFieldWithValueDto>>();
|
||
if (accountFieldCache == null)
|
||
{
|
||
// _segContext.Context.Span.AddLog(LogEvent.Message("【服务商账户】没能命中缓存，重新构建"));
|
||
var serviceProvider = cacheContext.TargetServiceProvider;
|
||
var accountId = await _interfaceForwardQuery.GetAccountIdAsync(serviceProvider.Id, _logRepository.AppId);
|
||
// 服务商授权
|
||
if (!serviceProvider.IsDisableAuth && accountId == default)
|
||
{
|
||
throw new BusinessException(message: $"服务商：{serviceProvider.Name} 下不存在此账户，请联系管理员或前往接口通创建");
|
||
}
|
||
|
||
accountFieldCache = await _interfaceForwardQuery.GetAccountFieldListAsync(accountId);
|
||
var ttl = await HandleAuthInterfaceAsync(cacheContext, accountFieldCache);
|
||
|
||
await RedisHelper.Client.SetAsync(accountCacheKey, accountFieldCache.ToJson(), ttl);
|
||
}
|
||
|
||
// _segContext.Context.Span.AddLog(LogEvent.Message("【服务商账户】构建完成"));
|
||
|
||
// 组装转发上下文
|
||
var res = new List<ForwardCoreContext>();
|
||
|
||
foreach (var item in cacheContext.TargetInterfaces)
|
||
{
|
||
item.TargetInterface.Timeout ??= cacheContext.TargetServiceProvider.Timeout;
|
||
var context = new ForwardCoreContext
|
||
{
|
||
TargetInterface = item.TargetInterface,
|
||
FixedFieldList = item.FixedFieldList,
|
||
AccountFieldList = accountFieldCache,
|
||
ReturnConfigList = item.ReturnConfigList,
|
||
FormFieldList = item.FormFieldList,
|
||
InParamTreeList = item.InParamTreeList,
|
||
PrefixScript = item.PrefixScript,
|
||
PostfixScript = item.PostfixScript
|
||
};
|
||
var token = context.FixedFieldList.FirstOrDefault(x => x.FieldValue == GlobalConst.Token);
|
||
if (token != default)
|
||
{
|
||
token.FieldValue = context.GetAccountFieldValueOrNull(GlobalConst.Token);
|
||
}
|
||
|
||
context.ServiceProviderRequestLog = new RequestLogDto(_logRepository.AppRequestLog.RequestId);
|
||
context.ServiceProviderRequestLog.Name = cacheContext.TargetServiceProvider.Name;
|
||
context.ServiceProviderRequestLog.Address = context.TargetInterface.RequestAddress;
|
||
context.ServiceProviderRequestLog.InterfaceCode = context.TargetInterface.Code;
|
||
context.ServiceProviderRequestLog.InterfaceName = context.TargetInterface.Name;
|
||
|
||
var flow = GetForwardFlow(context.TargetInterface.FlowCode, cacheContext.TargetServiceProvider.FlowCode);
|
||
if (flow == default)
|
||
throw new BusinessException(message: "找不到接口转发流程" + context.TargetInterface.FlowCode);
|
||
_flows.Add(context.TargetInterface.Code, flow);
|
||
|
||
|
||
res.Add(context);
|
||
}
|
||
|
||
// tips：把出参映射树放到最后一个上下文中，是为了在流程结束时执行出参映射
|
||
res.Last().OutParamTreeList = cacheContext.OutParamTreeList;
|
||
|
||
// _segContext.Context.Span.AddLog(LogEvent.Message("【转发上下文】全部组装完成"));
|
||
|
||
return res;
|
||
}
|
||
|
||
// 优先用服务商接口指定流程，其次使用服务商的，再其次是默认的
|
||
public IForwardFlow GetForwardFlow(string interfaceFlowCode, string serviceProviderFlowCode)
|
||
{
|
||
var flowCode = interfaceFlowCode.IsNullOrWhiteSpace()
|
||
? serviceProviderFlowCode.IsNullOrWhiteSpace()
|
||
? nameof(DefaultForwardFlow)
|
||
: serviceProviderFlowCode
|
||
: interfaceFlowCode;
|
||
return _serviceProvider.GetServices<IForwardFlow>()
|
||
.FirstOrDefault(x => x.GetType().Name == flowCode);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 处理服务商授权接口，返回一个过期时间表示服务商账户缓存的有效期
|
||
/// tips：因为我们将授权接口的返回值作为服务商账户字段的一部分一起缓存，所以这里需要返回过期时间
|
||
/// </summary>
|
||
/// <param name="cacheContext"></param>
|
||
/// <param name="accountFieldCache"></param>
|
||
/// <returns></returns>
|
||
/// <exception cref="ArgumentOutOfRangeException"></exception>
|
||
public virtual async Task<int> HandleAuthInterfaceAsync(ForwardCoreContextCache cacheContext,
|
||
List<ServiceProviderAccountFieldWithValueDto> accountFieldCache)
|
||
{
|
||
var interfaceDto = cacheContext.ServiceProviderAuthInterface;
|
||
|
||
if (interfaceDto == default)
|
||
{
|
||
return ContextCacheTtl;
|
||
}
|
||
|
||
var configs = cacheContext.ServiceProviderAuthConfigs;
|
||
|
||
// 构建入参
|
||
var body = new JObject();
|
||
foreach (var item in configs.Where(x => x.Flag == AuthParameterFlag.InParameter))
|
||
{
|
||
body[item.ParameterAlias] = item.Value1;
|
||
}
|
||
|
||
// 构建授权上下文
|
||
var authContext = new ForwardCoreContext
|
||
{
|
||
TargetInterface = interfaceDto.TargetInterface,
|
||
FixedFieldList = interfaceDto.FixedFieldList, // can filter service provider fix fields
|
||
AccountFieldList = accountFieldCache, // can add from cache
|
||
ReturnConfigList = interfaceDto.ReturnConfigList,
|
||
InParamTreeList = null,
|
||
OutParamTreeList = null,
|
||
PrefixScript = null,
|
||
PostfixScript = null,
|
||
OriginalInterfaceInput = body,
|
||
FormFieldList = interfaceDto.FormFieldList // normal case is null
|
||
};
|
||
authContext.ServiceProviderRequestLog = new RequestLogDto(_logRepository.AppRequestLog.RequestId)
|
||
{
|
||
Name = cacheContext.TargetServiceProvider.Name,
|
||
Address = authContext.TargetInterface.RequestAddress,
|
||
InterfaceCode = authContext.TargetInterface.Code,
|
||
InterfaceName = authContext.TargetInterface.Name
|
||
};
|
||
|
||
// 走动态脚本
|
||
var scriptConfig = configs.FirstOrDefault(x => x.Flag == AuthParameterFlag.Script);
|
||
if (scriptConfig != default && !scriptConfig.Value1.IsNullOrEmpty())
|
||
{
|
||
// 使用Natasha构建动态表达式
|
||
return NDelegate.RandomDomain().Func<ForwardCoreContext, int>(scriptConfig.Value1)(authContext);
|
||
}
|
||
|
||
// 授权也走主流程
|
||
if (!cacheContext.TargetServiceProvider.AuthInterfaceCode.IsNullOrWhiteSpace())
|
||
{
|
||
var flow = GetForwardFlow(interfaceDto.TargetInterface.FlowCode,
|
||
cacheContext.TargetServiceProvider.FlowCode);
|
||
if (flow == default)
|
||
throw new BusinessException(message: "找不到接口转发流程" + interfaceDto.TargetInterface.FlowCode);
|
||
_flows.Add(interfaceDto.TargetInterface.Code, flow);
|
||
}
|
||
|
||
// 请求授权接口
|
||
var response = await InternalForwardAsync(authContext);
|
||
|
||
// 添加到固定字段以便标准转发流程使用
|
||
var tokenConfig = configs.FirstOrDefault(x => x.Flag == AuthParameterFlag.Token);
|
||
if (tokenConfig != default)
|
||
{
|
||
accountFieldCache.Add(new ServiceProviderAccountFieldWithValueDto
|
||
{
|
||
FieldName = GlobalConst.Token,
|
||
FieldValue = tokenConfig.Value1.IsNullOrWhiteSpace()
|
||
? response[tokenConfig.ParameterAlias]?.ToString()
|
||
: tokenConfig.Value1.Trim() + " " + response[tokenConfig.ParameterAlias]
|
||
});
|
||
}
|
||
|
||
// 计算过期时间
|
||
var fixedEffectTimeConfig = configs.FirstOrDefault(x => x.Flag == AuthParameterFlag.FixedEffectTime);
|
||
int.TryParse(fixedEffectTimeConfig?.Value1, out var ttl);
|
||
var effectTimeConfig = configs.FirstOrDefault(x => x.Flag == AuthParameterFlag.EffectTime);
|
||
if (effectTimeConfig == default)
|
||
{
|
||
return ttl;
|
||
}
|
||
|
||
if (int.TryParse(response[effectTimeConfig.ParameterAlias]?.ToString(), out var expiredValue))
|
||
{
|
||
ttl = Enum.Parse<EffectTimeUnit>(effectTimeConfig.Value1) switch
|
||
{
|
||
EffectTimeUnit.Second => expiredValue,
|
||
EffectTimeUnit.Minute => expiredValue * 60,
|
||
EffectTimeUnit.Hour => expiredValue * 60 * 60,
|
||
_ => throw new ArgumentOutOfRangeException()
|
||
};
|
||
}
|
||
|
||
return ttl;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 获取上下文redis key
|
||
/// </summary>
|
||
/// <param name="upStreamCode">系统接口code</param>
|
||
/// <param name="serviceProviderCode">服务商code</param>
|
||
/// <returns></returns>
|
||
public static string GetContextCacheKey(string upStreamCode, string serviceProviderCode)
|
||
{
|
||
return $"ForwardContext:up_{upStreamCode}:sp_{serviceProviderCode}";
|
||
}
|
||
|
||
/// <summary>
|
||
/// 获取账户redis key
|
||
/// ex：
|
||
/// 如果appId==null，则返回：ForwardAccount:sp_{serviceProviderCode}:app_*
|
||
/// 如果appId!=null，则返回：ForwardAccount:sp_{serviceProviderCode}:app_{appId}
|
||
/// </summary>
|
||
/// <param name="serviceProviderCode">服务商code</param>
|
||
/// <param name="appId">应用id</param>
|
||
/// <returns></returns>
|
||
public static string GetAccountCacheKey(string serviceProviderCode, int? appId = null)
|
||
{
|
||
return $"ForwardAccount:sp_{serviceProviderCode}:app_{(appId == null ? "*" : appId)}";
|
||
}
|
||
|
||
public async Task<object> InternalForward3Async(List<ForwardCoreContext> contextList, List<RequestLogSummary> logs)
|
||
{
|
||
JToken ans = new JObject();
|
||
// 服务商接口X入参={系统入参数+服务商接口x-1返回参数+服务商接口x-2返回参数+..}
|
||
var input = contextList.First().OriginalInterfaceInput;
|
||
var retry = false;
|
||
for (var i = 0; i < contextList.Count; i++)
|
||
{
|
||
JToken res;
|
||
if (retry)
|
||
{
|
||
// 发起真实请求
|
||
res = await InternalForwardAsync(contextList[i], true);
|
||
}
|
||
else
|
||
{
|
||
var log = logs.FirstOrDefault(x => x.InterfaceCode == contextList[i].TargetInterface.Code);
|
||
// 如果是成功的，直接取日志内容
|
||
if (log is { IsSuccess: true })
|
||
{
|
||
res = JToken.Parse(log.Response);
|
||
}
|
||
else
|
||
{
|
||
res = await InternalForwardAsync(contextList[i], true);
|
||
// 1.当第一次真正发起请求，后面的都需要重试了，解决参数依赖问题。
|
||
// 2.流程变更问题，即以当前缓存中的流程为准，而非历史流程。
|
||
retry = true;
|
||
}
|
||
}
|
||
|
||
ans[contextList[i].TargetInterface.Code] = res;
|
||
|
||
if (i + 1 < contextList.Count)
|
||
{
|
||
// 合并请求结果作为下一个请求的入参
|
||
input[contextList[i].TargetInterface.Code] = res;
|
||
contextList[i + 1].OriginalInterfaceInput = input;
|
||
}
|
||
}
|
||
|
||
// 最后做一次反参映射，而不是每次
|
||
var context = contextList.Last();
|
||
return await _flows[context.TargetInterface.Code].OutParamRestructure(context, ans);
|
||
}
|
||
|
||
public async Task<JToken> InternalForward2Async(List<ForwardCoreContext> contextList)
|
||
{
|
||
JToken ans = new JObject();
|
||
var input = contextList.First().OriginalInterfaceInput;
|
||
for (var i = 0; i < contextList.Count; i++)
|
||
{
|
||
var res = await InternalForwardAsync(contextList[i], true);
|
||
ans[contextList[i].TargetInterface.Code] = res;
|
||
if (i + 1 < contextList.Count)
|
||
{
|
||
// 并入返回内容作为下一次请求的入参部分
|
||
input[contextList[i].TargetInterface.Code] = res;
|
||
contextList[i + 1].OriginalInterfaceInput = input;
|
||
}
|
||
}
|
||
|
||
// 最后做一次反参映射，而不是每次
|
||
var context = contextList.Last();
|
||
return await _flows[context.TargetInterface.Code].OutParamRestructure(context, ans);
|
||
}
|
||
|
||
public async Task<JToken> InternalForwardAsync(ForwardCoreContext context, bool isMulti = false)
|
||
{
|
||
try
|
||
{
|
||
var flow = _flows[context.TargetInterface.Code];
|
||
// _segContext.Context.Span.AddLog(LogEvent.Message("开始入参重构"));
|
||
context.OriginalInterfaceMappedInput = await flow.InParamRestructure(context);
|
||
|
||
// _segContext.Context.Span.AddLog(LogEvent.Message("* 开始请求"));
|
||
context.Feature.ServiceProviderDefaultQps = _interfaceRelayOptions.ServiceProviderDefaultQps;
|
||
context.Feature.SpinPeriod = _interfaceRelayOptions.SpinPeriod;
|
||
var response = await flow.ForwardAsync(context);
|
||
// _segContext.Context.Span.AddLog(LogEvent.Message("* 请求结束"));
|
||
|
||
if (isMulti) return response; // 一对多不在这映射
|
||
var res = await flow.OutParamRestructure(context, response);
|
||
|
||
// _segContext.Context.Span.AddLog(LogEvent.Message("完成反参重构"));
|
||
|
||
return res;
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
context.ServiceProviderRequestLog.IsSuccess = false;
|
||
context.ServiceProviderRequestLog.Exception = ex.ToString();
|
||
ExceptionDispatchInfo.Capture(ex).Throw();
|
||
|
||
return default;
|
||
}
|
||
finally
|
||
{
|
||
// 关联当前服务商日志
|
||
_logRepository.ServiceProviderRequestLog = context.ServiceProviderRequestLog;
|
||
// 写服务商日志
|
||
await _logRepository.AddLogAsync(context.ServiceProviderRequestLog);
|
||
}
|
||
}
|
||
|
||
public List<JObject> ObjectToObjectArray(object data)
|
||
{
|
||
if (JToken.FromObject(data) is not JArray jarray)
|
||
{
|
||
throw new BusinessException(message: "非法结构，请传对象数组");
|
||
}
|
||
|
||
var objectArray = new List<JObject>(jarray.Count);
|
||
foreach (var item in jarray) //数组对象校验
|
||
{
|
||
if (item is not JObject jobect)
|
||
{
|
||
throw new BusinessException(message: "批量处理必须传递对象数组");
|
||
}
|
||
|
||
objectArray.Add(jobect);
|
||
}
|
||
|
||
if (objectArray.Count < 1)
|
||
{
|
||
throw new BusinessException(message: "请勿传递空数组");
|
||
}
|
||
|
||
return objectArray;
|
||
}
|
||
} |