MiFaZhan
530 lines
17 KiB
Go
530 lines
17 KiB
Go
package renewchange
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"strings"
|
||
|
||
"github.com/gogf/gf/v2/database/gdb"
|
||
"github.com/gogf/gf/v2/encoding/gjson"
|
||
"github.com/gogf/gf/v2/errors/gerror"
|
||
"github.com/gogf/gf/v2/os/gtime"
|
||
|
||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consts"
|
||
)
|
||
|
||
// Repository 提供对续费/变更相关表的 typed 访问。
|
||
// 表名/列名都是常量（见 tables.go），与 Admin 同步链路保持一致。
|
||
type Repository struct{}
|
||
|
||
func New() *Repository {
|
||
return &Repository{}
|
||
}
|
||
|
||
const simCardRenewChangeFields = "id,iccid,msisdn,imsi,dept_id,current_package_code,package_status,outbound_batch_no,flow_pkg_begin_at,flow_pkg_end_at,tenant_id,created_dept,updated_by,updated_at"
|
||
|
||
func (r *Repository) database() gdb.DB {
|
||
return db()
|
||
}
|
||
|
||
// DB 返回底层 gdb.DB，供 service 层发起事务。
|
||
func (r *Repository) DB() gdb.DB {
|
||
return r.database()
|
||
}
|
||
|
||
// LoadBatchMain 按 batchNo+batchType 加载。
|
||
func (r *Repository) LoadBatchMain(ctx context.Context, batchNo string, batchType int) (*BatchMainRow, error) {
|
||
if batchNo == "" {
|
||
return nil, gerror.New("batch number is empty")
|
||
}
|
||
m := r.database().Model(TableBatchMain).Safe().Ctx(ctx).
|
||
Where("batch_no", batchNo).
|
||
Where("batch_type", batchType)
|
||
var out BatchMainRow
|
||
if err := m.Scan(&out); err != nil {
|
||
return nil, gerror.Wrap(err, "query batch main failed")
|
||
}
|
||
if out.Id == 0 {
|
||
return nil, nil
|
||
}
|
||
return &out, nil
|
||
}
|
||
|
||
// LoadBatchDetails 加载指定批次下指定状态集合的明细。
|
||
func (r *Repository) LoadBatchDetails(ctx context.Context, batchNo string, batchType int, statuses []int) ([]*BatchDetailRow, error) {
|
||
if batchNo == "" {
|
||
return nil, nil
|
||
}
|
||
m := r.database().Model(TableBatchDetail).Safe().Ctx(ctx).
|
||
Where("batch_no", batchNo).
|
||
Where("batch_type", batchType).
|
||
OrderAsc("row_no")
|
||
if len(statuses) > 0 {
|
||
m = m.WhereIn("detail_status", statuses)
|
||
}
|
||
var out []*BatchDetailRow
|
||
if err := m.Scan(&out); err != nil {
|
||
return nil, gerror.Wrap(err, "query batch details failed")
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
func (r *Repository) LoadCardsByIdentityValues(ctx context.Context, tenantID string, values []string) ([]*SimCardInfo, error) {
|
||
uniq := uniqueStrings(values)
|
||
if len(uniq) == 0 {
|
||
return nil, nil
|
||
}
|
||
tenantID = strings.TrimSpace(tenantID)
|
||
byID := map[int64]*SimCardInfo{}
|
||
for _, column := range []string{"iccid", "msisdn", "imsi"} {
|
||
var rows []*SimCardInfo
|
||
query := r.database().Model(TableSimCardInfo).Safe().Ctx(ctx).
|
||
Fields(simCardRenewChangeFields).
|
||
WhereIn(column, uniq)
|
||
if tenantID != "" {
|
||
query = query.Where("tenant_id", tenantID)
|
||
}
|
||
if err := query.Scan(&rows); err != nil {
|
||
return nil, gerror.Wrap(err, "query sim cards by identity failed")
|
||
}
|
||
for _, row := range rows {
|
||
if row != nil && row.Id > 0 {
|
||
byID[row.Id] = row
|
||
}
|
||
}
|
||
}
|
||
out := make([]*SimCardInfo, 0, len(byID))
|
||
for _, row := range byID {
|
||
out = append(out, row)
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// UpdateBatchMainStatus 在事务内更新主表。
|
||
func (r *Repository) UpdateBatchMainStatus(ctx context.Context, tx gdb.TX, batchNo string, batchType int, data *BatchMainRow) error {
|
||
updateData := map[string]any{
|
||
"batch_type": data.BatchType,
|
||
"batch_async_status": data.BatchAsyncStatus,
|
||
"run_result": data.RunResult,
|
||
"total_count": data.TotalCount,
|
||
"valid_count": data.ValidCount,
|
||
"success_count": data.SuccessCount,
|
||
"fail_count": data.FailCount,
|
||
"fail_reason": data.FailReason,
|
||
"batch_info": data.BatchInfo,
|
||
"end_time": data.EndTime,
|
||
"updated_by": data.UpdatedBy,
|
||
"updated_at": gtime.Now(),
|
||
"order_sn": data.OrderSn,
|
||
"target_dept_id": data.TargetDeptId,
|
||
}
|
||
if data.StartTime != nil && !data.StartTime.IsZero() {
|
||
updateData["start_time"] = data.StartTime
|
||
}
|
||
_, err := tx.Ctx(ctx).Model(TableBatchMain).
|
||
Where("batch_no", batchNo).
|
||
Where("batch_type", batchType).
|
||
Data(updateData).
|
||
Update()
|
||
return gerror.Wrap(err, "update batch main failed")
|
||
}
|
||
|
||
// ConvertBatchDetails 将指定状态的明细转换为目标类型与状态。
|
||
func (r *Repository) ConvertBatchDetails(ctx context.Context, tx gdb.TX, batchNo string, fromTypes []int, toType, toStatus int, data map[string]any) (int64, error) {
|
||
if data == nil {
|
||
data = map[string]any{}
|
||
}
|
||
data["batch_type"] = toType
|
||
data["detail_status"] = toStatus
|
||
data["updated_at"] = gtime.Now()
|
||
result, err := tx.Ctx(ctx).Model(TableBatchDetail).
|
||
Where("batch_no", batchNo).
|
||
WhereIn("batch_type", fromTypes).
|
||
Data(data).
|
||
Update()
|
||
if err != nil {
|
||
return 0, gerror.Wrap(err, "convert batch details failed")
|
||
}
|
||
affected, err := result.RowsAffected()
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return affected, nil
|
||
}
|
||
|
||
// UpdateBatchDetailSnapshot 覆盖指定明细的 snapshot 与状态。
|
||
func (r *Repository) UpdateBatchDetailSnapshot(ctx context.Context, tx gdb.TX, batchNo string, batchType, rowNo int, status int, snapshot *gjson.Json, reason string, userID int64) error {
|
||
data := map[string]any{
|
||
"detail_status": status,
|
||
"snapshot": snapshot,
|
||
"updated_at": gtime.Now(),
|
||
"updated_by": userID,
|
||
}
|
||
if reason != "" {
|
||
data["error_reason"] = reason
|
||
}
|
||
_, err := tx.Ctx(ctx).Model(TableBatchDetail).
|
||
Where("batch_no", batchNo).
|
||
Where("batch_type", batchType).
|
||
Where("row_no", rowNo).
|
||
Data(data).
|
||
Update()
|
||
return gerror.Wrap(err, "update batch detail snapshot failed")
|
||
}
|
||
|
||
// UpsertBatchDetail 按 (batch_no, batch_type, row_no) 写入明细（覆盖式）。
|
||
func (r *Repository) UpsertBatchDetail(ctx context.Context, tx gdb.TX, detail *BatchDetailRow) error {
|
||
_, err := tx.Ctx(ctx).Model(TableBatchDetail).
|
||
Where("batch_no", detail.BatchNo).
|
||
Where("batch_type", detail.BatchType).
|
||
Where("row_no", detail.RowNo).
|
||
Data(map[string]any{
|
||
"detail_status": detail.DetailStatus,
|
||
"blank_out_status": detail.BlankOutStatus,
|
||
"item_id": detail.ItemId,
|
||
"error_code": detail.ErrorCode,
|
||
"error_reason": detail.ErrorReason,
|
||
"snapshot": detail.Snapshot,
|
||
"updated_at": gtime.Now(),
|
||
"updated_by": detail.UpdatedBy,
|
||
}).
|
||
Update()
|
||
if err != nil {
|
||
return gerror.Wrap(err, "update batch detail failed")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// LoadCardsByIccids 加载正式卡。
|
||
func (r *Repository) LoadCardsByIccids(ctx context.Context, iccids []string) (map[string]*SimCardInfo, error) {
|
||
out := make(map[string]*SimCardInfo, len(iccids))
|
||
uniq := uniqueStrings(iccids)
|
||
if len(uniq) == 0 {
|
||
return out, nil
|
||
}
|
||
var rows []*SimCardInfo
|
||
if err := r.database().Model(TableSimCardInfo).Safe().Ctx(ctx).
|
||
Fields(simCardRenewChangeFields).
|
||
WhereIn("iccid", uniq).
|
||
Scan(&rows); err != nil {
|
||
return nil, gerror.Wrap(err, "query sim cards failed")
|
||
}
|
||
for _, row := range rows {
|
||
if row == nil {
|
||
continue
|
||
}
|
||
row.Iccid = strings.TrimSpace(row.Iccid)
|
||
if row.Iccid == "" {
|
||
continue
|
||
}
|
||
out[row.Iccid] = row
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// LockCardsByIccids 在事务内按 ICCID 锁行加载正式卡。
|
||
func (r *Repository) LockCardsByIccids(ctx context.Context, tx gdb.TX, iccids []string) (map[string]*SimCardInfo, error) {
|
||
out := make(map[string]*SimCardInfo, len(iccids))
|
||
uniq := uniqueStrings(iccids)
|
||
if len(uniq) == 0 {
|
||
return out, nil
|
||
}
|
||
var rows []*SimCardInfo
|
||
if err := tx.Ctx(ctx).Model(TableSimCardInfo).
|
||
Fields(simCardRenewChangeFields).
|
||
WhereIn("iccid", uniq).
|
||
LockUpdate().
|
||
Scan(&rows); err != nil {
|
||
return nil, gerror.Wrap(err, "lock sim cards failed")
|
||
}
|
||
for _, row := range rows {
|
||
if row == nil {
|
||
continue
|
||
}
|
||
row.Iccid = strings.TrimSpace(row.Iccid)
|
||
if row.Iccid == "" {
|
||
continue
|
||
}
|
||
out[row.Iccid] = row
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// LoadActiveBindRowsByIccids loads every pending/effective formal package
|
||
// subscription for prepare validation. Add-on package rows (103/104) are
|
||
// intentionally excluded from current-subscription selection.
|
||
func (r *Repository) LoadActiveBindRowsByIccids(ctx context.Context, iccids []string) ([]*SimCardBindPkg, error) {
|
||
uniq := uniqueStrings(iccids)
|
||
if len(uniq) == 0 {
|
||
return nil, nil
|
||
}
|
||
var rows []*SimCardBindPkg
|
||
if err := r.database().Ctx(ctx).Model(TableSimCardBindPkg).
|
||
WhereIn("iccid", uniq).
|
||
WhereIn("package_type", consts.FormalPackageTypes()).
|
||
WhereIn("effective_status", []int{consts.EffectiveStatusPending, consts.EffectiveStatusEffective}).
|
||
OrderDesc("effective_status").
|
||
OrderAsc("effective_time").
|
||
OrderAsc("id").
|
||
Scan(&rows); err != nil {
|
||
return nil, gerror.Wrap(err, "load active binds failed")
|
||
}
|
||
return rows, nil
|
||
}
|
||
|
||
// LoadRenewBindRowsByIccids includes expired formal-package history so renewal
|
||
// can restart after a card has naturally expired. Add-on package rows (103/104)
|
||
// never participate in renewal selection.
|
||
func (r *Repository) LoadRenewBindRowsByIccids(ctx context.Context, iccids []string) ([]*SimCardBindPkg, error) {
|
||
uniq := uniqueStrings(iccids)
|
||
if len(uniq) == 0 {
|
||
return nil, nil
|
||
}
|
||
var rows []*SimCardBindPkg
|
||
if err := r.database().Ctx(ctx).Model(TableSimCardBindPkg).
|
||
WhereIn("iccid", uniq).
|
||
WhereIn("package_type", consts.FormalPackageTypes()).
|
||
WhereIn("effective_status", []int{
|
||
consts.EffectiveStatusPending,
|
||
consts.EffectiveStatusEffective,
|
||
consts.EffectiveStatusExpired,
|
||
}).
|
||
OrderDesc("effective_status").
|
||
OrderAsc("effective_time").
|
||
OrderAsc("id").
|
||
Scan(&rows); err != nil {
|
||
return nil, gerror.Wrap(err, "load renew binds failed")
|
||
}
|
||
return rows, nil
|
||
}
|
||
|
||
// LockActiveBindRowsByIccids is the transactional counterpart used by change
|
||
// execution before any business write. It locks formal package rows only.
|
||
func (r *Repository) LockActiveBindRowsByIccids(ctx context.Context, tx gdb.TX, iccids []string) ([]*SimCardBindPkg, error) {
|
||
uniq := uniqueStrings(iccids)
|
||
if len(uniq) == 0 {
|
||
return nil, nil
|
||
}
|
||
var rows []*SimCardBindPkg
|
||
if err := tx.Ctx(ctx).Model(TableSimCardBindPkg).
|
||
WhereIn("iccid", uniq).
|
||
WhereIn("package_type", consts.FormalPackageTypes()).
|
||
WhereIn("effective_status", []int{consts.EffectiveStatusPending, consts.EffectiveStatusEffective}).
|
||
OrderDesc("effective_status").
|
||
OrderAsc("effective_time").
|
||
OrderAsc("id").
|
||
LockUpdate().
|
||
Scan(&rows); err != nil {
|
||
return nil, gerror.Wrap(err, "lock active binds failed")
|
||
}
|
||
return rows, nil
|
||
}
|
||
|
||
// LockRenewBindRowsByIccids includes expired formal-package history for renewal
|
||
// execution. Change execution keeps using LockActiveBindRowsByIccids.
|
||
func (r *Repository) LockRenewBindRowsByIccids(ctx context.Context, tx gdb.TX, iccids []string) ([]*SimCardBindPkg, error) {
|
||
uniq := uniqueStrings(iccids)
|
||
if len(uniq) == 0 {
|
||
return nil, nil
|
||
}
|
||
var rows []*SimCardBindPkg
|
||
if err := tx.Ctx(ctx).Model(TableSimCardBindPkg).
|
||
WhereIn("iccid", uniq).
|
||
WhereIn("package_type", consts.FormalPackageTypes()).
|
||
WhereIn("effective_status", []int{
|
||
consts.EffectiveStatusPending,
|
||
consts.EffectiveStatusEffective,
|
||
consts.EffectiveStatusExpired,
|
||
}).
|
||
OrderDesc("effective_status").
|
||
OrderAsc("effective_time").
|
||
OrderAsc("id").
|
||
LockUpdate().
|
||
Scan(&rows); err != nil {
|
||
return nil, gerror.Wrap(err, "lock renew binds failed")
|
||
}
|
||
return rows, nil
|
||
}
|
||
|
||
// LockPendingBindsByIccids 锁行加载 pending 状态的未来订购。
|
||
func (r *Repository) LockPendingBindsByIccids(ctx context.Context, tx gdb.TX, iccids []string) ([]*SimCardBindPkg, error) {
|
||
uniq := uniqueStrings(iccids)
|
||
if len(uniq) == 0 {
|
||
return nil, nil
|
||
}
|
||
var rows []*SimCardBindPkg
|
||
if err := tx.Ctx(ctx).Model(TableSimCardBindPkg).
|
||
WhereIn("iccid", uniq).
|
||
WhereIn("package_type", consts.FormalPackageTypes()).
|
||
Where("effective_status", consts.EffectiveStatusPending).
|
||
OrderAsc("effective_time").
|
||
OrderAsc("id").
|
||
LockUpdate().
|
||
Scan(&rows); err != nil {
|
||
return nil, gerror.Wrap(err, "lock pending binds failed")
|
||
}
|
||
return rows, nil
|
||
}
|
||
|
||
// DeletePendingBindsBySubscriptionNos 删除（作废）一批订阅号。
|
||
func (r *Repository) DeletePendingBindsBySubscriptionNos(ctx context.Context, tx gdb.TX, subscriptionNos []string) (int64, error) {
|
||
uniq := uniqueStrings(subscriptionNos)
|
||
if len(uniq) == 0 {
|
||
return 0, nil
|
||
}
|
||
result, err := tx.Ctx(ctx).Model(TableSimCardBindPkg).
|
||
WhereIn("subscription_no", uniq).
|
||
Delete()
|
||
if err != nil {
|
||
return 0, gerror.Wrap(err, "delete pending binds failed")
|
||
}
|
||
affected, err := result.RowsAffected()
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return affected, nil
|
||
}
|
||
|
||
// LoadPackageByCode 按 packageCode 读取套餐。
|
||
func (r *Repository) LoadPackageByCode(ctx context.Context, tenantID, packageCode string) (*FlowPackageInfo, error) {
|
||
if packageCode == "" {
|
||
return nil, errPkgNotFound
|
||
}
|
||
var row FlowPackageInfo
|
||
err := r.database().Model(TableSimFlowPackageInfo).Safe().Ctx(ctx).
|
||
Where("tenant_id", tenantID).
|
||
Where("package_code", strings.TrimSpace(packageCode)).
|
||
Scan(&row)
|
||
if err != nil {
|
||
return nil, gerror.Wrap(err, "query package failed")
|
||
}
|
||
if row.Id == 0 {
|
||
return nil, errPkgNotFound
|
||
}
|
||
return &row, nil
|
||
}
|
||
|
||
// LoadDeptName 按 deptId 取客户名。
|
||
func (r *Repository) LoadDeptName(ctx context.Context, tenantID string, deptID int64) (string, error) {
|
||
row, err := r.LoadDept(ctx, tenantID, deptID)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
return strings.TrimSpace(row.DeptName), nil
|
||
}
|
||
|
||
// LoadDept returns the current department and its parent. Renewals use the
|
||
// parent department as the price owner for the current card owner.
|
||
func (r *Repository) LoadDept(ctx context.Context, tenantID string, deptID int64) (*SysDeptRow, error) {
|
||
if deptID <= 0 {
|
||
return nil, errDeptNotFound
|
||
}
|
||
var row SysDeptRow
|
||
err := r.database().Model(TableSysDept).Safe().Ctx(ctx).
|
||
Where("tenant_id", tenantID).
|
||
Where("dept_id", deptID).
|
||
Scan(&row)
|
||
if err != nil {
|
||
return nil, gerror.Wrap(err, "query dept failed")
|
||
}
|
||
if row.DeptId == 0 {
|
||
return nil, errDeptNotFound
|
||
}
|
||
return &row, nil
|
||
}
|
||
|
||
// LoadPriceOverride 读取代理续费/变更价格覆盖。
|
||
func (r *Repository) LoadPriceOverride(ctx context.Context, tenantID string, deptID, targetAgentID int64, scopeType int, packageCode string) (*FlowPackagePrice, error) {
|
||
var row FlowPackagePrice
|
||
err := r.database().Model(TableSimFlowPackagePrice).Safe().Ctx(ctx).
|
||
Where("tenant_id", tenantID).
|
||
Where("dept_id", deptID).
|
||
Where("agent_scope_type", scopeType).
|
||
Where("target_agent_id", targetAgentID).
|
||
Where("package_code", strings.TrimSpace(packageCode)).
|
||
Where("enabled", 1).
|
||
Scan(&row)
|
||
return normalizePriceOverrideResult(&row, err)
|
||
}
|
||
|
||
func normalizePriceOverrideResult(row *FlowPackagePrice, err error) (*FlowPackagePrice, error) {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return nil, nil
|
||
}
|
||
if err != nil {
|
||
return nil, gerror.Wrap(err, "query price override failed")
|
||
}
|
||
if row == nil || row.Id == 0 {
|
||
return nil, nil
|
||
}
|
||
return row, nil
|
||
}
|
||
|
||
// UpdateCardPackageFields 续费/变更后更新卡的当前套餐与流量状态。
|
||
func (r *Repository) UpdateCardPackageFields(ctx context.Context, tx gdb.TX, iccid string, fields map[string]any, userID int64) error {
|
||
if iccid == "" {
|
||
return gerror.New("iccid is empty")
|
||
}
|
||
fields["updated_by"] = userID
|
||
fields["updated_at"] = gtime.Now()
|
||
_, err := tx.Ctx(ctx).Model(TableSimCardInfo).
|
||
Where("iccid", iccid).
|
||
Data(fields).
|
||
Update()
|
||
return gerror.Wrap(err, "update sim card failed")
|
||
}
|
||
|
||
// InsertOrder 插入订单（返回 orderSn）。
|
||
// InsertOrderItem 插入订单项。
|
||
// UpdateOrderReturnStatus 更新订单退订状态。
|
||
// UpdateOrderItemExtra 合并并写入订单项 extraSkuData（JSON 字符串）。
|
||
// ActiveBindCountByOrderCodes 统计订单当前未作废的订购数。
|
||
func (r *Repository) ActiveBindCountByOrderCodes(ctx context.Context, tx gdb.TX, orderSNs []string) (map[string]int, error) {
|
||
out := make(map[string]int, len(orderSNs))
|
||
uniq := uniqueStrings(orderSNs)
|
||
if len(uniq) == 0 {
|
||
return out, nil
|
||
}
|
||
type row struct {
|
||
OrderCode string `orm:"order_code"`
|
||
Count int `orm:"count"`
|
||
}
|
||
var rows []row
|
||
if err := tx.Ctx(ctx).Model(TableSimCardBindPkg).
|
||
Fields("order_code", "COUNT(1) AS count").
|
||
WhereIn("order_code", uniq).
|
||
Group("order_code").
|
||
Scan(&rows); err != nil {
|
||
return nil, gerror.Wrap(err, "count active binds failed")
|
||
}
|
||
for _, r := range rows {
|
||
out[strings.TrimSpace(r.OrderCode)] = r.Count
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// InsertBind 插入新的订购记录。
|
||
func (r *Repository) InsertBind(ctx context.Context, tx gdb.TX, bind *SimCardBindPkg) error {
|
||
_, err := tx.Ctx(ctx).Model(TableSimCardBindPkg).Data(bind).Insert()
|
||
return gerror.Wrap(err, "insert bind failed")
|
||
}
|
||
|
||
// InsertSysOss 插入 OSS 记录，返回 ID。
|
||
func (r *Repository) InsertSysOss(ctx context.Context, tx gdb.TX, row *SysOssRow) (int64, error) {
|
||
id, err := tx.Ctx(ctx).Model(TableSysOss).Data(row).InsertAndGetId()
|
||
if err != nil {
|
||
return 0, gerror.Wrap(err, "insert sys oss failed")
|
||
}
|
||
return id, nil
|
||
}
|
||
|
||
// InsertBatchDownload 插入 t_resource_batch_download 记录，返回 ID。
|
||
func (r *Repository) InsertBatchDownload(ctx context.Context, tx gdb.TX, row *ResourceBatchDownloadRow) (int64, error) {
|
||
id, err := tx.Ctx(ctx).Model(TableTResourceBatchDl).Data(row).InsertAndGetId()
|
||
if err != nil {
|
||
return 0, gerror.Wrap(err, "insert batch download failed")
|
||
}
|
||
return id, nil
|
||
}
|