junhong_cmp_fiber/internal/infrastructure/packageexpiry/reminder_publisher.go

// Package packageexpiry 提供套餐临期提醒的 PostgreSQL 与 Outbox Adapter。
package packageexpiry

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"gorm.io/gorm"

	notificationapp "github.com/break/junhong_cmp_fiber/internal/application/notification"
	"github.com/break/junhong_cmp_fiber/internal/infrastructure/messaging/outbox"
	"github.com/break/junhong_cmp_fiber/internal/model/dto"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
	"github.com/break/junhong_cmp_fiber/pkg/errors"
)

var shanghaiLocation = time.FixedZone("Asia/Shanghai", 8*60*60)

// ReminderPublisher 将临期节点事实转换为个人客户通知 Outbox。
type ReminderPublisher struct {
	db     *gorm.DB
	outbox *outbox.Repository
}

// NewReminderPublisher 创建套餐临期通知 Outbox Adapter。
func NewReminderPublisher(db *gorm.DB, repository *outbox.Repository) *ReminderPublisher {
	return &ReminderPublisher{db: db, outbox: repository}
}

type recipientRow struct {
	AssetID    uint
	CustomerID uint
}

// Publish 批量解析资产关联个人客户，并在同一事务内幂等追加通知事件。
func (p *ReminderPublisher) Publish(ctx context.Context, candidates []dto.ExpiringAssetItem) error {
	if p == nil || p.db == nil || p.outbox == nil {
		return errors.New(errors.CodeInternalError, "套餐临期通知 Publisher 未配置")
	}
	recipients, err := p.loadRecipients(ctx, candidates)
	if err != nil {
		return err
	}
	return p.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		for _, candidate := range candidates {
			if candidate.DaysUntilFinalExpiry == nil || candidate.EstimatedFinalExpiresAt == nil {
				continue
			}
			key := recipientKey(candidate.AssetType, candidate.AssetID)
			for customerID := range recipients[key] {
				if err := p.appendNotification(ctx, tx, candidate, customerID); err != nil {
					return err
				}
			}
		}
		return nil
	})
}

func (p *ReminderPublisher) loadRecipients(ctx context.Context, candidates []dto.ExpiringAssetItem) (map[string]map[uint]struct{}, error) {
	cardIDs := make([]uint, 0)
	deviceIDs := make([]uint, 0)
	for _, candidate := range candidates {
		if candidate.AssetType == constants.AssetTypeIotCard {
			cardIDs = append(cardIDs, candidate.AssetID)
		} else if candidate.AssetType == constants.AssetTypeDevice {
			deviceIDs = append(deviceIDs, candidate.AssetID)
		}
	}
	result := make(map[string]map[uint]struct{})
	queries := []struct {
		assetType string
		ids       []uint
		sql       string
	}{
		{constants.AssetTypeIotCard, cardIDs, `
			SELECT c.id AS asset_id, b.customer_id
			FROM tb_iot_card c
			JOIN tb_personal_customer_device b ON c.virtual_no <> '' AND b.virtual_no = c.virtual_no
			WHERE c.id IN ? AND c.deleted_at IS NULL AND b.deleted_at IS NULL AND b.status = 1`},
		{constants.AssetTypeIotCard, cardIDs, `
			SELECT c.id AS asset_id, b.customer_id
			FROM tb_iot_card c
			JOIN tb_personal_customer_iccid b ON c.virtual_no = '' AND (b.iccid = c.iccid OR (b.iccid_19 <> '' AND b.iccid_19 = c.iccid_19))
			WHERE c.id IN ? AND c.deleted_at IS NULL AND b.deleted_at IS NULL AND b.status = 1`},
		{constants.AssetTypeDevice, deviceIDs, `
			SELECT d.id AS asset_id, b.customer_id
			FROM tb_device d
			JOIN tb_personal_customer_device b ON b.virtual_no = CASE WHEN d.virtual_no <> '' THEN d.virtual_no ELSE d.imei END
			WHERE d.id IN ? AND d.deleted_at IS NULL AND b.deleted_at IS NULL AND b.status = 1`},
	}
	for _, query := range queries {
		if len(query.ids) == 0 {
			continue
		}
		var rows []recipientRow
		if err := p.db.WithContext(ctx).Raw(query.sql, query.ids).Scan(&rows).Error; err != nil {
			return nil, errors.Wrap(errors.CodeDatabaseError, err, "批量查询临期资产关联客户失败")
		}
		for _, row := range rows {
			key := recipientKey(query.assetType, row.AssetID)
			if result[key] == nil {
				result[key] = make(map[uint]struct{})
			}
			result[key][row.CustomerID] = struct{}{}
		}
	}
	return result, nil
}

func (p *ReminderPublisher) appendNotification(ctx context.Context, tx *gorm.DB, candidate dto.ExpiringAssetItem, customerID uint) error {
	node := *candidate.DaysUntilFinalExpiry
	expiryDate := candidate.EstimatedFinalExpiresAt.In(shanghaiLocation).Format("20060102")
	assetCode := "c"
	if candidate.AssetType == constants.AssetTypeDevice {
		assetCode = "d"
	}
	eventID := fmt.Sprintf("pex:%s:%d:%s:%d:%d", assetCode, candidate.AssetID, expiryDate, node, customerID)
	assetID := strconv.FormatUint(uint64(candidate.AssetID), 10)
	_, err := p.outbox.AppendIdempotent(ctx, tx, outbox.Envelope{
		EventID: eventID, EventType: constants.OutboxEventTypePersonalCustomerDirectNotification,
		PayloadVersion: constants.NotificationPayloadVersionV1,
		AggregateType:  "package_expiry", AggregateID: strconv.FormatUint(uint64(candidate.PackageUsageID), 10),
		ResourceType: candidate.AssetType, ResourceID: assetID, BusinessKey: eventID,
		Payload: notificationapp.PersonalCustomerDirectPayload{
			RecipientID: customerID, NotificationType: constants.NotificationTypePackageExpiring,
			TemplateData: map[string]string{}, RefType: constants.NotificationRefTypeAsset,
			RefID: assetID, RefKey: candidate.Identifier, ExpiresAt: candidate.EstimatedFinalExpiresAt,
		},
	})
	if err != nil {
		return errors.Wrap(errors.CodeDatabaseError, err, "写入套餐临期通知事件失败")
	}
	return nil
}

func recipientKey(assetType string, assetID uint) string {
	return fmt.Sprintf("%s:%d", assetType, assetID)
}