forked from xiongyuxing/tiku-backend.net
feat: add referral crm queue endpoints
This commit is contained in:
xiong
@@ -71,6 +71,7 @@ public static class DependencyInjection
|
||||
services.AddScoped<ICommerceAdminService, CommerceAdminService>();
|
||||
services.AddScoped<IPointService, PointService>();
|
||||
services.AddScoped<IReferralService, ReferralService>();
|
||||
services.AddScoped<ICrmService, CrmService>();
|
||||
services.AddScoped<ITenantSecretService, TenantSecretService>();
|
||||
services.AddScoped<IPaymentProviderConfigService, PaymentProviderConfigService>();
|
||||
services.AddScoped<IPaymentProviderGateway, PaymentProviderGateway>();
|
||||
|
||||
420
Tiku.Infrastructure/Growth/CrmService.cs
Normal file
420
Tiku.Infrastructure/Growth/CrmService.cs
Normal file
@@ -0,0 +1,420 @@
|
||||
using System.Text.Json;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Tiku.Application.Growth;
|
||||
using Tiku.Domain.Common;
|
||||
using Tiku.Domain.Growth;
|
||||
using Tiku.Domain.Tenancy;
|
||||
using Tiku.Infrastructure.Persistence;
|
||||
|
||||
namespace Tiku.Infrastructure.Growth;
|
||||
|
||||
public sealed class CrmService(TikuDbContext dbContext) : ICrmService
|
||||
{
|
||||
private static readonly HashSet<string> SensitiveKeys = new(StringComparer.OrdinalIgnoreCase)
|
||||
{
|
||||
"secret",
|
||||
"password",
|
||||
"token",
|
||||
"accessToken",
|
||||
"refreshToken",
|
||||
"signature",
|
||||
"privateKey",
|
||||
"apiKey"
|
||||
};
|
||||
|
||||
public async Task<CrmConfigItem> GetConfigAsync(
|
||||
CrmAdminActor actor,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
await AssertAdminAsync(actor, cancellationToken);
|
||||
var config = await dbContext.CrmConfigs
|
||||
.AsNoTracking()
|
||||
.FirstOrDefaultAsync(item => item.TenantId == actor.TenantId, cancellationToken);
|
||||
return ToConfigItem(config ?? new CrmConfig { TenantId = actor.TenantId });
|
||||
}
|
||||
|
||||
public async Task<CrmConfigItem> UpsertConfigAsync(
|
||||
CrmAdminActor actor,
|
||||
UpsertCrmConfigCommand command,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
await AssertAdminAsync(actor, cancellationToken);
|
||||
var secretRef = NormalizeOptional(command.SecretRef);
|
||||
if (!string.IsNullOrWhiteSpace(command.Secret))
|
||||
{
|
||||
secretRef ??= "tenant_secrets:crm:webhook:default";
|
||||
await UpsertSecretAsync(actor.TenantId, secretRef, command.Secret, cancellationToken);
|
||||
}
|
||||
|
||||
var config = await dbContext.CrmConfigs
|
||||
.FirstOrDefaultAsync(item => item.TenantId == actor.TenantId, cancellationToken);
|
||||
if (config is null)
|
||||
{
|
||||
config = new CrmConfig { TenantId = actor.TenantId };
|
||||
dbContext.CrmConfigs.Add(config);
|
||||
}
|
||||
|
||||
config.Enabled = command.Enabled;
|
||||
config.Url = NormalizeOptional(command.Url);
|
||||
config.SecretRef = secretRef;
|
||||
config.FormName = NormalizeOptional(command.FormName);
|
||||
config.ExamType = NormalizeOptional(command.ExamType);
|
||||
config.TimeoutSeconds = command.TimeoutSeconds;
|
||||
config.DelaySeconds = command.DelaySeconds;
|
||||
config.AssignmentMode = ParseEnum(command.AssignmentMode, ReferralAssignmentMode.None, "invalid_assignment_mode");
|
||||
config.AssignmentPool = EnsureArray(command.AssignmentPool);
|
||||
config.AssignmentConfig = EnsureObject(command.AssignmentConfig);
|
||||
|
||||
await dbContext.SaveChangesAsync(cancellationToken);
|
||||
return ToConfigItem(config);
|
||||
}
|
||||
|
||||
public async Task<CrmList<CrmQueueItem>> GetQueueAsync(
|
||||
CrmAdminActor actor,
|
||||
CrmQueueQuery query,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
await AssertAdminAsync(actor, cancellationToken);
|
||||
var items = ApplyQueueQuery(actor.TenantId, query);
|
||||
var result = await items
|
||||
.OrderByDescending(item => item.CreatedAt)
|
||||
.Take(Math.Clamp(query.Limit ?? 100, 1, 500))
|
||||
.ToArrayAsync(cancellationToken);
|
||||
return new CrmList<CrmQueueItem>(result.Select(ToQueueItem).ToArray());
|
||||
}
|
||||
|
||||
public async Task<CrmDeadLetterResult> GetDeadLettersAsync(
|
||||
CrmAdminActor actor,
|
||||
CrmQueueQuery query,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
await AssertAdminAsync(actor, cancellationToken);
|
||||
var deadLetters = dbContext.CrmWebhookQueue
|
||||
.AsNoTracking()
|
||||
.Where(item =>
|
||||
item.TenantId == actor.TenantId &&
|
||||
(item.Status == CrmWebhookQueueStatus.Failed ||
|
||||
item.Status == CrmWebhookQueueStatus.Discarded));
|
||||
if (!string.IsNullOrWhiteSpace(query.Source))
|
||||
{
|
||||
var source = query.Source.Trim();
|
||||
deadLetters = deadLetters.Where(item => item.Source == source);
|
||||
}
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(query.Status))
|
||||
{
|
||||
deadLetters = deadLetters.Where(item => item.Status == ParseEnum(query.Status, CrmWebhookQueueStatus.Failed, "invalid_crm_queue_status"));
|
||||
}
|
||||
|
||||
var items = await deadLetters
|
||||
.OrderBy(item => item.CreatedAt)
|
||||
.Take(Math.Clamp(query.Limit ?? 100, 1, 500))
|
||||
.ToArrayAsync(cancellationToken);
|
||||
var failed = await dbContext.CrmWebhookQueue.CountAsync(
|
||||
item => item.TenantId == actor.TenantId && item.Status == CrmWebhookQueueStatus.Failed,
|
||||
cancellationToken);
|
||||
var discarded = await dbContext.CrmWebhookQueue.CountAsync(
|
||||
item => item.TenantId == actor.TenantId && item.Status == CrmWebhookQueueStatus.Discarded,
|
||||
cancellationToken);
|
||||
var oldest = await dbContext.CrmWebhookQueue
|
||||
.Where(item => item.TenantId == actor.TenantId && item.Status == CrmWebhookQueueStatus.Failed)
|
||||
.OrderBy(item => item.CreatedAt)
|
||||
.Select(item => (DateTimeOffset?)item.CreatedAt)
|
||||
.FirstOrDefaultAsync(cancellationToken);
|
||||
|
||||
return new CrmDeadLetterResult(
|
||||
new CrmDeadLetterSummary(failed, discarded, failed + discarded, oldest),
|
||||
items.Select(ToQueueItem).ToArray());
|
||||
}
|
||||
|
||||
public async Task<CrmList<CrmQueueLogItem>> GetLogsAsync(
|
||||
CrmAdminActor actor,
|
||||
CrmQueueLogQuery query,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
await AssertAdminAsync(actor, cancellationToken);
|
||||
var logs = dbContext.CrmWebhookLogs
|
||||
.AsNoTracking()
|
||||
.Where(item => item.TenantId == actor.TenantId);
|
||||
if (query.QueueId.HasValue)
|
||||
{
|
||||
var queue = await dbContext.CrmWebhookQueue
|
||||
.AsNoTracking()
|
||||
.Where(item => item.TenantId == actor.TenantId && item.Id == query.QueueId.Value)
|
||||
.Select(item => new { item.Id, item.RecordId })
|
||||
.FirstOrDefaultAsync(cancellationToken)
|
||||
?? throw new CrmException("CRM queue item was not found.", "crm_queue_not_found");
|
||||
logs = logs.Where(item => item.RecordId == queue.RecordId || item.RecordId == queue.Id.ToString());
|
||||
}
|
||||
|
||||
var result = await logs
|
||||
.OrderByDescending(item => item.CreatedAt)
|
||||
.Take(Math.Clamp(query.Limit ?? 50, 1, 200))
|
||||
.ToArrayAsync(cancellationToken);
|
||||
return new CrmList<CrmQueueLogItem>(result.Select(ToLogItem).ToArray());
|
||||
}
|
||||
|
||||
public async Task<CrmQueueItem> ApplyQueueActionAsync(
|
||||
CrmAdminActor actor,
|
||||
CrmQueueActionCommand command,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
await AssertAdminAsync(actor, cancellationToken);
|
||||
var item = await dbContext.CrmWebhookQueue
|
||||
.FirstOrDefaultAsync(entry => entry.TenantId == actor.TenantId && entry.Id == command.QueueId, cancellationToken)
|
||||
?? throw new CrmException("CRM queue item was not found.", "crm_queue_not_found");
|
||||
var action = NormalizeOptional(command.Action) ?? "retry";
|
||||
if (action.Equals("retry", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
if (item.Status is not (CrmWebhookQueueStatus.Failed or CrmWebhookQueueStatus.Discarded or CrmWebhookQueueStatus.Retrying))
|
||||
{
|
||||
throw new CrmException("CRM queue item cannot be retried.", "crm_queue_status_invalid");
|
||||
}
|
||||
|
||||
item.Status = CrmWebhookQueueStatus.Pending;
|
||||
item.NextAttemptAt = DateTimeOffset.UtcNow;
|
||||
item.LastError = null;
|
||||
}
|
||||
else if (action.Equals("ignore", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
if (item.Status is CrmWebhookQueueStatus.Sent)
|
||||
{
|
||||
throw new CrmException("CRM queue item cannot be ignored.", "crm_queue_status_invalid");
|
||||
}
|
||||
|
||||
item.Status = CrmWebhookQueueStatus.Discarded;
|
||||
item.LastError = NormalizeOptional(command.Note) ?? item.LastError;
|
||||
}
|
||||
else
|
||||
{
|
||||
throw new CrmException("CRM queue action was invalid.", "invalid_crm_queue_action");
|
||||
}
|
||||
|
||||
dbContext.CrmWebhookLogs.Add(new CrmWebhookLog
|
||||
{
|
||||
TenantId = actor.TenantId,
|
||||
RecordId = item.RecordId ?? item.Id.ToString(),
|
||||
LeadId = item.LeadId,
|
||||
Outcome = "operator_action",
|
||||
ErrorMessage = item.LastError,
|
||||
RequestPayload = Redact(command.Metadata ?? JsonSerializer.SerializeToElement(new { note = command.Note })),
|
||||
ResponseSummary = action,
|
||||
Attempt = item.Attempts
|
||||
});
|
||||
await dbContext.SaveChangesAsync(cancellationToken);
|
||||
return ToQueueItem(item);
|
||||
}
|
||||
|
||||
private IQueryable<CrmWebhookQueueItem> ApplyQueueQuery(Guid tenantId, CrmQueueQuery query)
|
||||
{
|
||||
var items = dbContext.CrmWebhookQueue
|
||||
.AsNoTracking()
|
||||
.Where(item => item.TenantId == tenantId);
|
||||
if (query.QueueId.HasValue)
|
||||
{
|
||||
items = items.Where(item => item.Id == query.QueueId.Value);
|
||||
}
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(query.Status))
|
||||
{
|
||||
items = items.Where(item => item.Status == ParseEnum(query.Status, CrmWebhookQueueStatus.Pending, "invalid_crm_queue_status"));
|
||||
}
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(query.Source))
|
||||
{
|
||||
var source = query.Source.Trim();
|
||||
items = items.Where(item => item.Source == source);
|
||||
}
|
||||
|
||||
return items;
|
||||
}
|
||||
|
||||
private async Task UpsertSecretAsync(
|
||||
Guid tenantId,
|
||||
string secretRef,
|
||||
string secret,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var item = await dbContext.TenantSecrets
|
||||
.FirstOrDefaultAsync(secretItem => secretItem.TenantId == tenantId && secretItem.SecretRef == secretRef, cancellationToken);
|
||||
if (item is null)
|
||||
{
|
||||
item = new TenantSecret
|
||||
{
|
||||
TenantId = tenantId,
|
||||
Purpose = "crm",
|
||||
Provider = "webhook",
|
||||
SecretKey = "default",
|
||||
SecretRef = secretRef
|
||||
};
|
||||
dbContext.TenantSecrets.Add(item);
|
||||
}
|
||||
else
|
||||
{
|
||||
item.RotatedAt = DateTimeOffset.UtcNow;
|
||||
}
|
||||
|
||||
item.Status = TenantSecretStatus.Active;
|
||||
item.SecretPayload = JsonSerializer.SerializeToElement(new { webhookSecret = secret });
|
||||
}
|
||||
|
||||
private async Task AssertAdminAsync(CrmAdminActor actor, CancellationToken cancellationToken)
|
||||
{
|
||||
var isAdmin = await dbContext.TenantMemberships.AnyAsync(
|
||||
item =>
|
||||
item.TenantId == actor.TenantId &&
|
||||
item.UserId == actor.UserId &&
|
||||
item.Status == MembershipStatus.Active &&
|
||||
(item.Role == TenantRole.PlatformAdmin ||
|
||||
item.Role == TenantRole.TenantOwner ||
|
||||
item.Role == TenantRole.TenantAdmin),
|
||||
cancellationToken);
|
||||
if (!isAdmin)
|
||||
{
|
||||
throw new CrmException("CRM admin access was denied.", "crm_access_denied");
|
||||
}
|
||||
}
|
||||
|
||||
private static CrmConfigItem ToConfigItem(CrmConfig item)
|
||||
{
|
||||
return new CrmConfigItem(
|
||||
item.Id,
|
||||
item.Enabled,
|
||||
item.Url,
|
||||
item.SecretRef,
|
||||
item.AssignmentMode.ToString(),
|
||||
item.AssignmentPool,
|
||||
item.FormName,
|
||||
item.ExamType,
|
||||
item.TimeoutSeconds,
|
||||
item.DelaySeconds,
|
||||
item.UpdatedAt);
|
||||
}
|
||||
|
||||
private static CrmQueueItem ToQueueItem(CrmWebhookQueueItem item)
|
||||
{
|
||||
return new CrmQueueItem(
|
||||
item.Id,
|
||||
item.Status.ToString(),
|
||||
item.Attempts,
|
||||
item.LeadId,
|
||||
item.RecordId,
|
||||
item.Source,
|
||||
item.Provider,
|
||||
RedactText(item.LastError),
|
||||
item.ScheduledAt,
|
||||
item.NextAttemptAt,
|
||||
Redact(item.Payload));
|
||||
}
|
||||
|
||||
private static CrmQueueLogItem ToLogItem(CrmWebhookLog item)
|
||||
{
|
||||
return new CrmQueueLogItem(
|
||||
item.Id,
|
||||
null,
|
||||
item.RecordId,
|
||||
item.LeadId,
|
||||
item.HttpCode,
|
||||
item.Outcome ?? "unknown",
|
||||
RedactText(item.ErrorMessage),
|
||||
item.ResponseSummary,
|
||||
Redact(item.RequestPayload),
|
||||
RedactText(item.ResponseSummary),
|
||||
item.CreatedAt);
|
||||
}
|
||||
|
||||
private static JsonElement EnsureObject(JsonElement? value)
|
||||
{
|
||||
if (value is null || value.Value.ValueKind is JsonValueKind.Undefined or JsonValueKind.Null)
|
||||
{
|
||||
return JsonDefaults.Object();
|
||||
}
|
||||
|
||||
if (value.Value.ValueKind != JsonValueKind.Object)
|
||||
{
|
||||
throw new CrmException("CRM JSON value must be an object.", "invalid_json_payload");
|
||||
}
|
||||
|
||||
return value.Value.Clone();
|
||||
}
|
||||
|
||||
private static JsonElement EnsureArray(JsonElement? value)
|
||||
{
|
||||
if (value is null || value.Value.ValueKind is JsonValueKind.Undefined or JsonValueKind.Null)
|
||||
{
|
||||
return JsonDefaults.Array();
|
||||
}
|
||||
|
||||
if (value.Value.ValueKind != JsonValueKind.Array)
|
||||
{
|
||||
throw new CrmException("CRM JSON value must be an array.", "invalid_json_payload");
|
||||
}
|
||||
|
||||
return value.Value.Clone();
|
||||
}
|
||||
|
||||
private static JsonElement Redact(JsonElement value)
|
||||
{
|
||||
object? converted = RedactValue(value);
|
||||
return JsonSerializer.SerializeToElement(converted);
|
||||
}
|
||||
|
||||
private static object? RedactValue(JsonElement value)
|
||||
{
|
||||
return value.ValueKind switch
|
||||
{
|
||||
JsonValueKind.Object => value.EnumerateObject()
|
||||
.ToDictionary(
|
||||
property => property.Name,
|
||||
property => SensitiveKeys.Contains(property.Name) ? "***" : RedactValue(property.Value)),
|
||||
JsonValueKind.Array => value.EnumerateArray().Select(RedactValue).ToArray(),
|
||||
JsonValueKind.String => value.GetString(),
|
||||
JsonValueKind.Number when value.TryGetInt64(out var longValue) => longValue,
|
||||
JsonValueKind.Number when value.TryGetDecimal(out var decimalValue) => decimalValue,
|
||||
JsonValueKind.True => true,
|
||||
JsonValueKind.False => false,
|
||||
_ => null
|
||||
};
|
||||
}
|
||||
|
||||
private static string? RedactText(string? value)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(value))
|
||||
{
|
||||
return value;
|
||||
}
|
||||
|
||||
var result = value;
|
||||
foreach (var key in SensitiveKeys)
|
||||
{
|
||||
result = result.Replace(key, "***", StringComparison.OrdinalIgnoreCase);
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
private static TEnum ParseEnum<TEnum>(string? value, TEnum defaultValue, string errorCode)
|
||||
where TEnum : struct, Enum
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(value))
|
||||
{
|
||||
return defaultValue;
|
||||
}
|
||||
|
||||
var normalized = value.Replace("_", string.Empty, StringComparison.Ordinal);
|
||||
foreach (var enumValue in Enum.GetValues<TEnum>())
|
||||
{
|
||||
if (string.Equals(enumValue.ToString(), normalized, StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
return enumValue;
|
||||
}
|
||||
}
|
||||
|
||||
throw new CrmException("CRM enum value was invalid.", errorCode);
|
||||
}
|
||||
|
||||
private static string? NormalizeOptional(string? value)
|
||||
{
|
||||
return string.IsNullOrWhiteSpace(value) ? null : value.Trim();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user