break
c64f3d8b80
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 8m31s
251 lines
11 KiB
Go
251 lines
11 KiB
Go
package audit
|
||
|
||
import (
|
||
"context"
|
||
"strconv"
|
||
"time"
|
||
|
||
"gorm.io/datatypes"
|
||
"gorm.io/gorm"
|
||
|
||
auditinfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
retentionquery "github.com/break/junhong_cmp_fiber/internal/query/retention"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
)
|
||
|
||
var timelineRegistry = auditinfra.NewRegistry()
|
||
|
||
// ResourceSearchFilter 定义注册资源的精确标识搜索。
|
||
type ResourceSearchFilter struct {
|
||
ResourceType string
|
||
Keyword string
|
||
OnlineFrom time.Time
|
||
Page int
|
||
PageSize int
|
||
}
|
||
|
||
// ResourceSearchPage 是资源候选稳定分页结果。
|
||
type ResourceSearchPage struct {
|
||
Total int64 `json:"total" description:"符合精确标识的资源总数"`
|
||
Page int `json:"page" description:"当前页码"`
|
||
PageSize int `json:"page_size" description:"每页数量"`
|
||
Items []ResourceCandidate `json:"items" description:"当前资源或历史快照解析出的候选资源"`
|
||
Retention retentionquery.Info `json:"retention" description:"在线查询留存边界"`
|
||
}
|
||
|
||
// ResourceCandidate 是当前业务表或历史事件快照解析出的稳定资源候选。
|
||
type ResourceCandidate struct {
|
||
ResourceType string `json:"resource_type" enum:"iot_card,device,shop,order,refund" description:"资源类型稳定编码"`
|
||
ResourceID string `json:"resource_id" description:"资源内部稳定ID，可传给通用资源时间线接口"`
|
||
ResourceKey string `json:"resource_key" description:"资源业务稳定Key"`
|
||
DisplayName string `json:"display_name" description:"资源展示名称"`
|
||
IdentitySnapshot map[string]any `json:"identity_snapshot" description:"当前业务表或历史事件保存的资源身份快照"`
|
||
Historical bool `json:"historical" description:"是否仅由历史事件快照解析，true不代表资源当前仍存在"`
|
||
}
|
||
|
||
// ResourceTimelineFilter 定义通用资源时间线筛选。
|
||
type ResourceTimelineFilter struct {
|
||
ResourceType string
|
||
ResourceID string
|
||
CreatedFrom *time.Time
|
||
CreatedTo *time.Time
|
||
Action string
|
||
Result string
|
||
Page int
|
||
PageSize int
|
||
}
|
||
|
||
// SearchResources 按 Resource Registry 声明的稳定标识精确搜索资源。
|
||
func (q *Query) SearchResources(ctx context.Context, filter ResourceSearchFilter) (*ResourceSearchPage, error) {
|
||
if err := q.authorize(ctx); err != nil {
|
||
return nil, err
|
||
}
|
||
if filter.Keyword == "" || !searchableResourceType(filter.ResourceType) {
|
||
return nil, errors.New(errors.CodeInvalidParam)
|
||
}
|
||
retention, err := retentionquery.Load(ctx, q.db, retentionquery.SourceAudit)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
filter.OnlineFrom = retention.OnlineFrom
|
||
filter.Page, filter.PageSize = normalizePage(filter.Page, filter.PageSize)
|
||
items, total, err := q.searchCurrent(ctx, filter)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if total == 0 {
|
||
items, total, err = q.searchHistorical(ctx, filter)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
return &ResourceSearchPage{Total: total, Page: filter.Page, PageSize: filter.PageSize, Items: items, Retention: retention}, nil
|
||
}
|
||
|
||
// ResourceTimeline 查询注册资源作为任意关系参与的统一事件时间线。
|
||
func (q *Query) ResourceTimeline(ctx context.Context, filter ResourceTimelineFilter) (*EventPage, error) {
|
||
if filter.ResourceID == "" || !timelineResourceType(filter.ResourceType) {
|
||
return nil, errors.New(errors.CodeInvalidParam)
|
||
}
|
||
return q.List(ctx, EventFilter{
|
||
ResourceType: filter.ResourceType, ResourceID: filter.ResourceID,
|
||
CreatedFrom: filter.CreatedFrom, CreatedTo: filter.CreatedTo,
|
||
Action: filter.Action, Result: filter.Result,
|
||
Page: filter.Page, PageSize: filter.PageSize,
|
||
})
|
||
}
|
||
|
||
func timelineResourceType(resourceType string) bool {
|
||
_, ok := timelineRegistry.Resource(resourceType)
|
||
return ok
|
||
}
|
||
|
||
func (q *Query) searchCurrent(ctx context.Context, filter ResourceSearchFilter) ([]ResourceCandidate, int64, error) {
|
||
switch filter.ResourceType {
|
||
case constants.AuditResourceIotCard:
|
||
var rows []model.IotCard
|
||
query := q.db.WithContext(ctx).Where("iccid = ? OR virtual_no = ? OR iccid_19 = ? OR iccid_20 = ?", filter.Keyword, filter.Keyword, filter.Keyword, filter.Keyword)
|
||
return searchModels(query, filter, &rows, func(row model.IotCard) ResourceCandidate {
|
||
return candidate(filter.ResourceType, row.ID, row.ICCID, row.ICCID, map[string]any{
|
||
"id": row.ID, "iccid": row.ICCID, "virtual_no": row.VirtualNo, "msisdn": row.MSISDN,
|
||
"carrier_type": row.CarrierType, "shop_id": row.ShopID, "series_id": row.SeriesID, "generation": row.Generation,
|
||
})
|
||
})
|
||
case constants.AuditResourceDevice:
|
||
var rows []model.Device
|
||
query := q.db.WithContext(ctx).Where("virtual_no = ? OR imei = ? OR sn = ?", filter.Keyword, filter.Keyword, filter.Keyword)
|
||
return searchModels(query, filter, &rows, func(row model.Device) ResourceCandidate {
|
||
return candidate(filter.ResourceType, row.ID, deviceCandidateKey(row), deviceCandidateKey(row), map[string]any{
|
||
"id": row.ID, "virtual_no": row.VirtualNo, "imei": row.IMEI, "sn": row.SN,
|
||
"device_name": row.DeviceName, "device_model": row.DeviceModel, "shop_id": row.ShopID,
|
||
"series_id": row.SeriesID, "generation": row.Generation,
|
||
})
|
||
})
|
||
case constants.AuditResourceShop:
|
||
var rows []model.Shop
|
||
return searchModels(q.db.WithContext(ctx).Where("shop_code = ?", filter.Keyword), filter, &rows, func(row model.Shop) ResourceCandidate {
|
||
return candidate(filter.ResourceType, row.ID, row.ShopCode, row.ShopName, map[string]any{
|
||
"id": row.ID, "shop_code": row.ShopCode, "shop_name": row.ShopName, "parent_id": row.ParentID, "level": row.Level,
|
||
})
|
||
})
|
||
case constants.AuditResourceOrder:
|
||
var rows []model.Order
|
||
return searchModels(q.db.WithContext(ctx).Where("order_no = ?", filter.Keyword), filter, &rows, func(row model.Order) ResourceCandidate {
|
||
return candidate(filter.ResourceType, row.ID, row.OrderNo, row.OrderNo, map[string]any{
|
||
"id": row.ID, "order_no": row.OrderNo, "buyer_type": row.BuyerType, "buyer_id": row.BuyerID,
|
||
"asset_identifier": row.AssetIdentifier, "total_amount": row.TotalAmount,
|
||
"payment_method": row.PaymentMethod, "payment_status": row.PaymentStatus,
|
||
})
|
||
})
|
||
case constants.AuditResourceRefund:
|
||
var rows []model.RefundRequest
|
||
return searchModels(q.db.WithContext(ctx).Where("refund_no = ?", filter.Keyword), filter, &rows, func(row model.RefundRequest) ResourceCandidate {
|
||
return candidate(filter.ResourceType, row.ID, row.RefundNo, row.RefundNo, map[string]any{
|
||
"id": row.ID, "refund_no": row.RefundNo, "order_id": row.OrderID, "order_no": row.OrderNo,
|
||
"asset_identifier": row.AssetIdentifier, "shop_id": row.ShopID,
|
||
"requested_refund_amount": row.RequestedRefundAmount, "status": row.Status,
|
||
})
|
||
})
|
||
default:
|
||
return nil, 0, errors.New(errors.CodeInvalidParam)
|
||
}
|
||
}
|
||
|
||
func searchModels[T any](query *gorm.DB, filter ResourceSearchFilter, rows *[]T, project func(T) ResourceCandidate) ([]ResourceCandidate, int64, error) {
|
||
var total int64
|
||
if err := query.Model(new(T)).Count(&total).Error; err != nil {
|
||
return nil, 0, errors.Wrap(errors.CodeDatabaseError, err, "统计资源候选失败")
|
||
}
|
||
if err := query.Order("id ASC").Offset((filter.Page - 1) * filter.PageSize).Limit(filter.PageSize).Find(rows).Error; err != nil {
|
||
return nil, 0, errors.Wrap(errors.CodeDatabaseError, err, "查询资源候选失败")
|
||
}
|
||
items := make([]ResourceCandidate, 0, len(*rows))
|
||
for _, row := range *rows {
|
||
items = append(items, project(row))
|
||
}
|
||
return items, total, nil
|
||
}
|
||
|
||
func (q *Query) searchHistorical(ctx context.Context, filter ResourceSearchFilter) ([]ResourceCandidate, int64, error) {
|
||
base := q.historicalIdentifierQuery(ctx, filter)
|
||
var total int64
|
||
if err := base.Distinct("resource_id").Count(&total).Error; err != nil {
|
||
return nil, 0, errors.Wrap(errors.CodeDatabaseError, err, "统计历史资源候选失败")
|
||
}
|
||
latest := q.historicalIdentifierQuery(ctx, filter).
|
||
Select("DISTINCT ON (resource_id) resource_id, resource_key, display_name, identity_snapshot, created_at, id").
|
||
Order("resource_id ASC, created_at DESC, id DESC")
|
||
var rows []historicalResourceRow
|
||
if err := q.db.WithContext(ctx).Table("(?) AS historical", latest).
|
||
Order("resource_id ASC").Offset((filter.Page - 1) * filter.PageSize).Limit(filter.PageSize).Find(&rows).Error; err != nil {
|
||
return nil, 0, errors.Wrap(errors.CodeDatabaseError, err, "查询历史资源候选失败")
|
||
}
|
||
items := make([]ResourceCandidate, 0, len(rows))
|
||
for _, row := range rows {
|
||
identity, err := decodeObject(row.IdentitySnapshot)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
items = append(items, ResourceCandidate{
|
||
ResourceType: filter.ResourceType, ResourceID: row.ResourceID,
|
||
ResourceKey: row.ResourceKey, DisplayName: row.DisplayName,
|
||
IdentitySnapshot: identity, Historical: true,
|
||
})
|
||
}
|
||
return items, total, nil
|
||
}
|
||
|
||
func (q *Query) historicalIdentifierQuery(ctx context.Context, filter ResourceSearchFilter) *gorm.DB {
|
||
query := q.db.WithContext(ctx).Model(&model.AuditEventResource{}).
|
||
Where("resource_type = ? AND resource_id IS NOT NULL AND created_at >= ?", filter.ResourceType, filter.OnlineFrom.UTC())
|
||
switch filter.ResourceType {
|
||
case constants.AuditResourceIotCard:
|
||
return query.Where("resource_key = ? OR identity_snapshot ->> 'iccid' = ? OR identity_snapshot ->> 'iccid_19' = ? OR identity_snapshot ->> 'iccid_20' = ? OR identity_snapshot ->> 'virtual_no' = ?", filter.Keyword, filter.Keyword, filter.Keyword, filter.Keyword, filter.Keyword)
|
||
case constants.AuditResourceDevice:
|
||
return query.Where("resource_key = ? OR identity_snapshot ->> 'virtual_no' = ? OR identity_snapshot ->> 'imei' = ? OR identity_snapshot ->> 'sn' = ?", filter.Keyword, filter.Keyword, filter.Keyword, filter.Keyword)
|
||
case constants.AuditResourceShop:
|
||
return query.Where("resource_key = ? OR identity_snapshot ->> 'shop_code' = ?", filter.Keyword, filter.Keyword)
|
||
case constants.AuditResourceOrder:
|
||
return query.Where("resource_key = ? OR identity_snapshot ->> 'order_no' = ?", filter.Keyword, filter.Keyword)
|
||
case constants.AuditResourceRefund:
|
||
return query.Where("resource_key = ? OR identity_snapshot ->> 'refund_no' = ?", filter.Keyword, filter.Keyword)
|
||
default:
|
||
return query.Where("1 = 0")
|
||
}
|
||
}
|
||
|
||
type historicalResourceRow struct {
|
||
ResourceID string
|
||
ResourceKey string
|
||
DisplayName string
|
||
IdentitySnapshot datatypes.JSON
|
||
}
|
||
|
||
func candidate(resourceType string, id uint, key, name string, identity map[string]any) ResourceCandidate {
|
||
return ResourceCandidate{
|
||
ResourceType: resourceType, ResourceID: strconv.FormatUint(uint64(id), 10),
|
||
ResourceKey: key, DisplayName: name, IdentitySnapshot: identity,
|
||
}
|
||
}
|
||
|
||
func searchableResourceType(resourceType string) bool {
|
||
switch resourceType {
|
||
case constants.AuditResourceIotCard, constants.AuditResourceDevice, constants.AuditResourceShop,
|
||
constants.AuditResourceOrder, constants.AuditResourceRefund:
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
func deviceCandidateKey(row model.Device) string {
|
||
for _, value := range []string{row.VirtualNo, row.IMEI, row.SN} {
|
||
if value != "" {
|
||
return value
|
||
}
|
||
}
|
||
return strconv.FormatUint(uint64(row.ID), 10)
|
||
}
|