hyapp-server/services/activity-service/internal/app/action_confirm_mq.go

package app

import (
	"context"
	"encoding/json"
	"log/slog"
	"time"

	"hyapp/pkg/appcode"
	"hyapp/pkg/logx"
	"hyapp/pkg/messagemq"
	"hyapp/pkg/rocketmqx"
	"hyapp/pkg/usermq"
	"hyapp/services/activity-service/internal/config"
	messagedomain "hyapp/services/activity-service/internal/domain/message"
)

const userOutboxEventRoleInvitationCreated = "RoleInvitationCreated"

func newMessageActionUserOutboxConsumer(cfg config.Config, services *serviceBundle) (*rocketmqx.Consumer, error) {
	consumer, err := rocketmqx.NewConsumer(userOutboxMessageActionConsumerConfig(cfg.RocketMQ))
	if err != nil {
		return nil, err
	}
	if err := consumer.Subscribe(cfg.RocketMQ.UserOutbox.Topic, usermq.TagUserOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
		event, appCode, ok, err := roleInvitationConfirmEventFromUserMessage(message.Body)
		if err != nil || !ok {
			return err
		}
		_, _, err = services.actionConfirm.ConsumeRoleInvitationCreated(appcode.WithContext(ctx, appCode), event)
		return err
	}); err != nil {
		_ = consumer.Shutdown()
		return nil, err
	}
	return consumer, nil
}

type roleInvitationCreatedPayload struct {
	InvitationID       int64                                    `json:"invitation_id"`
	InvitationType     string                                   `json:"invitation_type"`
	InviterUserID      int64                                    `json:"inviter_user_id"`
	TargetUserID       int64                                    `json:"target_user_id"`
	AgencyName         string                                   `json:"agency_name"`
	Inviter            messagedomain.RoleInvitationUserSnapshot `json:"inviter"`
	Target             messagedomain.RoleInvitationUserSnapshot `json:"target"`
	CreatedAtMS        int64                                    `json:"created_at_ms"`
	ParentBDUserID     int64                                    `json:"parent_bd_user_id"`
	ParentLeaderUserID int64                                    `json:"parent_leader_user_id"`
}

func roleInvitationConfirmEventFromUserMessage(body []byte) (messagedomain.RoleInvitationConfirmEvent, string, bool, error) {
	message, err := usermq.DecodeUserOutboxMessage(body)
	if err != nil {
		return messagedomain.RoleInvitationConfirmEvent{}, "", false, err
	}
	if message.EventType != userOutboxEventRoleInvitationCreated {
		return messagedomain.RoleInvitationConfirmEvent{}, message.AppCode, false, nil
	}
	var payload roleInvitationCreatedPayload
	if err := json.Unmarshal([]byte(message.PayloadJSON), &payload); err != nil {
		return messagedomain.RoleInvitationConfirmEvent{}, message.AppCode, true, err
	}
	return messagedomain.RoleInvitationConfirmEvent{
		SourceEventID:      message.EventID,
		InvitationID:       payload.InvitationID,
		InvitationType:     payload.InvitationType,
		InviterUserID:      payload.InviterUserID,
		TargetUserID:       payload.TargetUserID,
		AgencyName:         payload.AgencyName,
		Inviter:            payload.Inviter,
		Target:             payload.Target,
		CreatedAtMS:        firstPositiveMS(payload.CreatedAtMS, message.OccurredAtMS),
		RawPayloadJSON:     message.PayloadJSON,
		ParentBDUserID:     payload.ParentBDUserID,
		ParentLeaderUserID: payload.ParentLeaderUserID,
	}, message.AppCode, true, nil
}

func (a *App) runMessageActionOutboxWorker(ctx context.Context) {
	options := a.messageActionWorkerOptions
	ticker := time.NewTicker(options.OutboxPollInterval)
	defer ticker.Stop()
	for {
		processed, err := a.processMessageActionOutboxBatch(ctx)
		if err != nil && ctx.Err() == nil {
			logx.Error(ctx, "message_action_outbox_publish_failed", err, slog.String("node_id", a.workerNodeID))
		}
		if processed >= options.OutboxBatchSize {
			continue
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

func (a *App) processMessageActionOutboxBatch(ctx context.Context) (int, error) {
	options := a.messageActionWorkerOptions
	workerCtx := appcode.WithContext(ctx, appcode.Default)
	records, err := a.actionConfirm.ClaimPendingOutbox(workerCtx, options.OutboxBatchSize, options.OutboxLockTTL)
	if err != nil {
		return 0, err
	}
	for _, record := range records {
		publishCtx, cancel := context.WithTimeout(context.Background(), options.PublishTimeout)
		err := a.publishMessageActionOutboxRecord(publishCtx, record)
		cancel()
		recordCtx := appcode.WithContext(context.Background(), record.AppCode)
		if err != nil {
			nextRetryAtMS := time.Now().UTC().Add(options.OutboxPollInterval).UnixMilli()
			if markErr := a.actionConfirm.MarkOutboxRetryable(recordCtx, record.EventID, err, nextRetryAtMS); markErr != nil {
				return len(records), markErr
			}
			return len(records), err
		}
		if err := a.actionConfirm.MarkOutboxDelivered(recordCtx, record.EventID); err != nil {
			return len(records), err
		}
	}
	return len(records), nil
}

func (a *App) publishMessageActionOutboxRecord(ctx context.Context, record messagedomain.ActionConfirmOutbox) error {
	body, err := messagemq.EncodeActionOutboxMessage(messagemq.ActionOutboxMessage{
		AppCode:      record.AppCode,
		EventID:      record.EventID,
		EventType:    record.EventType,
		ConfirmID:    record.ConfirmID,
		PayloadJSON:  record.PayloadJSON,
		OccurredAtMS: record.CreatedAtMS,
	})
	if err != nil {
		return err
	}
	return a.messageActionProducer.SendSync(ctx, rocketmqx.Message{
		Topic: a.messageActionOutboxTopic,
		Tag:   messagemq.TagMessageActionOutboxEvent,
		Keys:  []string{record.EventID, record.ConfirmID},
		Body:  body,
	})
}

func firstPositiveMS(values ...int64) int64 {
	for _, value := range values {
		if value > 0 {
			return value
		}
	}
	return time.Now().UTC().UnixMilli()
}