Method finalStatusUpdate

registry/services/asyncprocessing/service.go:342–383

nolint:nestif

(
	ctx context.Context,
	e *events.Event[*asyncprocessing.ExecuteAsyncTaskPayload],
	task *types.Task,
	processingErr error,
)

//nolint:nestif
func (s *Service) finalStatusUpdate(
	ctx context.Context,
	e *events.Event[*asyncprocessing.ExecuteAsyncTaskPayload],
	task *types.Task,
	processingErr error,
) (bool, error) {
	var runAgain bool
	err := s.tx.WithTx(
		ctx, func(ctx context.Context) error {
			_, err := s.taskRepository.LockForUpdate(ctx, task)
			if err != nil {
				log.Ctx(ctx).Error().Msgf("failed to lock task [%s] for update: %v", task.Key, err)
				return fmt.Errorf("failed to lock task for update: %w", err)
			}
			if processingErr != nil {
				log.Error().Ctx(ctx).Msgf("processing error for task [%s]: %v", task.Key, processingErr)
				err = s.taskSourceRepository.UpdateSourceStatus(ctx, e.ID, types.TaskStatusFailure, processingErr.Error())
				if err != nil {
					return err
				}
				runAgain, err = s.taskRepository.CompleteTask(ctx, task.Key, types.TaskStatusFailure)
				if err != nil {
					return err
				}
			} else {
				err = s.taskSourceRepository.UpdateSourceStatus(ctx, e.ID, types.TaskStatusSuccess, "")
				if err != nil {
					return err
				}
				runAgain, err = s.taskRepository.CompleteTask(ctx, task.Key, types.TaskStatusSuccess)
				if err != nil {
					return err
				}
			}
			return err
		})
	if err != nil {
		return false, fmt.Errorf("failed to update final statuses of task and sources, eventID:%s, task key: %s, err: %w",
			e.ID, task.Key, err)
	}
	return runAgain, nil
}

func (s *Service) ProcessingStatusUpdate(ctx context.Context, task *types.Task, runID string) error {
	err := s.tx.WithTx(