qaas это queue as a service то есть обертка кафки

unc (s *cpaRefundSubscriber) ProcessMessage(ctx context.Context, message any, meta qaasgen.Meta) error {
	msg, ok := message.(*cpa_refund.CpaRefund)
	if !ok {
		return qaas.ErrCantConvertMessage
	}

	streamEventRepo := s.dao.NewStreamEventQuery(ctx)
	streamEventFinishedRepo := s.dao.NewStreamEventFinishedQuery(ctx)

	events, err := s.dao.NewEventQuery().FindByPK(ctx, entity.EventPK{
		Purchase:   domain.PurchaseCPA.StringValue(),
		Kind:       domain.WriteOffKind.String(),
		ExternalID: msg.ActionId,
	})
	if err != nil {
		return err
	}
	var event domain.Event
	if len(events) > 0 {
		event, err = events[0].ToDomainEvent()
		if err != nil {
			return err
		}
	} else if msg.UserId != nil {
		holdKey := domain.NewHoldKey("cpa", *msg.UserId)
		events, err := streamEventRepo.GetByPK(entity.StreamEventPK{
			HoldKey:    holdKey.String(),
			Kind:       domain.WriteOffKind.String(),
			ExternalID: msg.ActionId,
		})
		if err != nil {
			return err
		}

		if len(events) == 0 {
			// событие признания выручки не было найдено, пропускаем возврат
			if meta.DeferCount() > cpaRefundNotFoundRetryLimit {
				slog.ErrorContext(ctx, "event not found, stopped to retry")
				return nil
			}

			return qaas.ErrToRetry
		}
		event, err = events[0].ToDomainEvent()
		if err != nil {
			return err
		}
	} else {
		// TODO: BILL-19245 вырезать вместе с msg.UserId != nil на 76 строке
		events, err := streamEventRepo.GetByExternalID(msg.ActionId)
		if err != nil {
			return err
		}

		if len(events) == 0 {
			// событие признания выручки не было найдено, пропускаем возврат
			if meta.DeferCount() > cpaRefundNotFoundRetryLimit {
				slog.ErrorContext(ctx, "event not found, stopped to retry")
				return nil
			}

			return qaas.ErrToRetry
		}
		for _, e := range events {
			if e.Kind == domain.WriteOffKind.String() {
				event, err = e.ToDomainEvent()
				if err != nil {
					return err
				}
				break
			}
		}
	}

	if msg.UserId != nil && *msg.UserId != event.UserID {
		slog.ErrorContext(ctx, "userId from message not equal userId from streamEventRepo")
	}

	if event.Purchase != domain.PurchaseCPA || event.Kind != domain.WriteOffKind {
		return errors.New("event not found")
	}
	// проверяем не обрабатывали ли мы уже это событие
	// (идемпотентность Create не поможет, если событие уже уехало в холодное хранилище)
	exists, err := streamEventFinishedRepo.Exists(event.HoldKey.String(), domain.RefundKind, msg.ActionId)
	if err != nil {
		return err
	}
	if exists {
		slog.WarnContext(ctx, "already processed, skip")
		return nil
	}

	// складываем события в табличку stream_events
	err = streamEventRepo.Create(
		entity.StreamEvent{
			HoldKey:      event.HoldKey.String(),
			ExternalID:   msg.ActionId,
			Kind:         domain.RefundKind.String(),
			StatusID:     domain.StatusNew.Int16Value(),
			ServiceID:    int16(domain.ServiceID_CPA),
			UserID:       event.UserID,
			AmountTotal:  event.Amount.Total,
			AmountRUB:    event.Amount.RUB,
			AmountBonus:  event.Amount.Bonus,
			MaxBonusRate: 0, // не используется в возвратах
			Context: entity.Context{
				VerticalID: event.Context.VerticalID,
			},
			CreatedAt: time.Now(),
			UpdatedAt: time.Now(),
		},
	)
	if err != nil {
		return err
	}

	streamEventMarker := s.dao.NewStreamEventMarkerQuery()
	exists, err = streamEventMarker.Exists(event.HoldKey.String())
	if err != nil {
		return err
	}
	if exists {
		return nil
	}

	// проверяем метку в редисе
	err = streamEventMarker.Set(event.HoldKey.String())
	if err != nil {
		return err
	}

	qaasMsg := start_preprocess.StartPreprocess{
		HoldKey: event.HoldKey.String(),
		UserID:  event.UserID,
		EventID: uuid.New().String(),
	}

	// ставим qaas задачу в отложенную очередь
	err = s.qaasClient.PublishStartPreprocess10s(ctx, []start_preprocess.StartPreprocess{qaasMsg})
	if err != nil {
		// снимаем метку, так как задача не была поставлена
		errDelete := streamEventMarker.Delete(event.HoldKey.String())
		return errors.Join(err, errDelete)
	}
	logformatter.LogPublishedMessage(ctx, StartPreprocessTopic, qaasMsg)

	return nil
}

объясни мне что тут происходит?