public MigrationResult run(Set<String> completedAlertConditions, Set<String> completedAlarmCallbacks) {
        final MigrationResult.Builder result = MigrationResult.builder();

        streamsCollection.find().forEach(stream -> {
            final String streamId = stream.getObjectId("_id").toHexString();
            final String streamTitle = stream.getString("title");

            final FindIterable<Document> iterable = alarmCallbacksCollection.find(Filters.eq("stream_id", streamId));
            final Set<NotificationDto> notifications = Streams.stream(iterable)
                    .map(alarmCallback -> {
                        final String callbackId = alarmCallback.getObjectId("_id").toHexString();

                        if (completedAlarmCallbacks.contains(callbackId)) {
                            result.addCompletedAlarmCallback(callbackId);
                            return dbNotificationService.get(callbackId).orElse(null);
                        }

                        try {
                            final NotificationDto notificationDto = migrateAlarmCallback(alarmCallback);
                            result.addCompletedAlarmCallback(callbackId);
                            return notificationDto;
                        } catch (Exception e) {
                            LOG.error("Couldn't migrate legacy alarm callback on stream <{}/{}>: {}", streamTitle, streamId, alarmCallback, e);
                            return null;
                        }
                    })
                    .filter(Objects::nonNull)
                    .collect(Collectors.toSet());

            if (!stream.containsKey("alert_conditions")) {
                return;
            }

            @SuppressWarnings("unchecked")
            final List<Document> list = (List<Document>) stream.get("alert_conditions");

            list.forEach(alertCondition -> {
                final String conditionId = alertCondition.getString("id");
                final String conditionType = alertCondition.getString("type");

                if (completedAlertConditions.contains(conditionId)) {
                    result.addCompletedAlertCondition(conditionId);
                    return;
                }
                try {
                    switch (conditionType) {
                        case "message_count":
                            migrateMessageCount(new Helper(stream, alertCondition, notifications));
                            result.addCompletedAlertCondition(conditionId);
                            break;
                        case "field_value":
                            migrateFieldValue(new Helper(stream, alertCondition, notifications));
                            result.addCompletedAlertCondition(conditionId);
                            break;
                        case "field_content_value":
                            migrateFieldContentValue(new Helper(stream, alertCondition, notifications));
                            result.addCompletedAlertCondition(conditionId);
                            break;
                        default:
                            LOG.warn("Couldn't migrate unknown legacy alert condition type: {}", conditionType);
                    }
                } catch (Exception e) {
                    LOG.error("Couldn't migrate legacy alert condition on stream <{}/{}>: {}", streamTitle, streamId, alertCondition, e);
                }
            });
        });

        return result.build();
    }
@Test
    @MongoDBFixtures("legacy-alert-conditions.json")
    public void runWithMigrationStatus() {
        final int migratedConditions = 9; // Only 8 because we pass one migrated condition in
        final int migratedCallbacks = 3;  // Only 2 because we pass one migrated callback in

        assertThat(migrator.run(Collections.singleton("00000000-0000-0000-0000-000000000002"), Collections.singleton("54e3deadbeefdeadbeef0001"))).satisfies(result -> {
            assertThat(result.completedAlertConditions()).containsOnly(
                    "00000000-0000-0000-0000-000000000001",
                    "00000000-0000-0000-0000-000000000002",
                    "00000000-0000-0000-0000-000000000003",
                    "00000000-0000-0000-0000-000000000004",
                    "00000000-0000-0000-0000-000000000005",
                    "00000000-0000-0000-0000-000000000006",
                    "00000000-0000-0000-0000-000000000007",
                    "00000000-0000-0000-0000-000000000008",
                    "00000000-0000-0000-0000-000000000009",
                    "00000000-0000-0000-0000-000000000010"
            );
            assertThat(result.completedAlarmCallbacks()).containsOnly(
                    "54e3deadbeefdeadbeef0001",
                    "54e3deadbeefdeadbeef0002",
                    "54e3deadbeefdeadbeef0003",
                    "54e3deadbeefdeadbeef0004"
            );
        });

        // Make sure we use the EventDefinitionHandler to create the event definitions
        verify(eventDefinitionHandler, times(migratedConditions)).create(any(EventDefinitionDto.class), any(Optional.class));

        // Make sure we use the NotificationResourceHandler to create the notifications
        verify(notificationResourceHandler, times(migratedCallbacks)).create(any(NotificationDto.class), any(Optional.class));

        assertThat(eventDefinitionService.streamAll().count()).isEqualTo(migratedConditions);
        assertThat(notificationService.streamAll().count()).isEqualTo(migratedCallbacks);
    }