<commit_before>/*
 * Copyright 2010-2014 Ning, Inc.
 * Copyright 2014 The Billing Project, LLC
 *
 * Ning licenses this file to you under the Apache License, version 2.0
 * (the "License"); you may not use this file except in compliance with the
 * License.  You may obtain a copy of the License at:
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */

package org.killbill.billing.plugin.analytics.reports.scheduler;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.UUID;

import javax.annotation.Nullable;

import org.joda.time.DateTime;
import org.killbill.billing.plugin.analytics.dao.BusinessDBIProvider;
import org.killbill.billing.plugin.analytics.reports.configuration.ReportsConfigurationModelDao;
import org.killbill.billing.plugin.analytics.reports.configuration.ReportsConfigurationModelDao.Frequency;
import org.killbill.clock.Clock;
import org.killbill.killbill.osgi.libs.killbill.OSGIKillbillDataSource;
import org.killbill.killbill.osgi.libs.killbill.OSGIKillbillLogService;
import org.killbill.notificationq.DefaultNotificationQueueService;
import org.killbill.notificationq.api.NotificationEvent;
import org.killbill.notificationq.api.NotificationEventWithMetadata;
import org.killbill.notificationq.api.NotificationQueue;
import org.killbill.notificationq.api.NotificationQueueService.NotificationQueueAlreadyExists;
import org.killbill.notificationq.api.NotificationQueueService.NotificationQueueHandler;
import org.osgi.service.log.LogService;
import org.skife.jdbi.v2.Call;
import org.skife.jdbi.v2.Handle;
import org.skife.jdbi.v2.IDBI;
import org.skife.jdbi.v2.sqlobject.mixins.Transmogrifier;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Objects;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;

import static org.killbill.billing.plugin.analytics.AnalyticsActivator.ANALYTICS_QUEUE_SERVICE;

public class JobsScheduler {

    // Current version of the jobs in the notification queue
    // This is useful to retrieve all currently scheduled ones
    private static final Long JOBS_SCHEDULER_VERSION = 1L;

    private final OSGIKillbillLogService logService;
    private final IDBI dbi;
    private final Clock clock;
    private final NotificationQueue jobQueue;

    public JobsScheduler(final OSGIKillbillLogService logService,
                         final OSGIKillbillDataSource osgiKillbillDataSource,
                         final Clock clock,
                         final DefaultNotificationQueueService notificationQueueService) throws NotificationQueueAlreadyExists {
        this.logService = logService;
        this.clock = clock;

        dbi = BusinessDBIProvider.get(osgiKillbillDataSource.getDataSource());
        final NotificationQueueHandler notificationQueueHandler = new NotificationQueueHandler() {

            @Override
            public void handleReadyNotification(final NotificationEvent eventJson, final DateTime eventDateTime, final UUID userToken, final Long searchKey1, final Long searchKey2) {
                if (eventJson == null || !(eventJson instanceof AnalyticsReportJob)) {
                    logService.log(LogService.LOG_ERROR, "Analytics report service received an unexpected event: " + eventJson);
                    return;
                }

                final AnalyticsReportJob job = (AnalyticsReportJob) eventJson;
                logService.log(LogService.LOG_INFO, "Starting job for " + job.getReportName());

                try {
                    callStoredProcedure(job.getRefreshProcedureName());
                } finally {
                    schedule(job, null);
                    logService.log(LogService.LOG_INFO, "Ending job for " + job.getReportName());
                }
            }
        };
        jobQueue = notificationQueueService.createNotificationQueue(ANALYTICS_QUEUE_SERVICE,
                                                                    "reports-jobs",
                                                                    notificationQueueHandler);
    }

    public void start() {
        jobQueue.startQueue();
    }

    public void shutdownNow() {
        jobQueue.stopQueue();
    }

    public void scheduleNow(final ReportsConfigurationModelDao report) {
        final AnalyticsReportJob eventJson = new AnalyticsReportJob(report);
        schedule(eventJson, clock.getUTCNow(), null);
    }

    public void schedule(final ReportsConfigurationModelDao report, final Transmogrifier transmogrifier) {
        final AnalyticsReportJob eventJson = new AnalyticsReportJob(report);
        schedule(eventJson, transmogrifier);
    }

    public void unSchedule(final ReportsConfigurationModelDao report, final Transmogrifier transmogrifier) {
        final AnalyticsReportJob eventJson = new AnalyticsReportJob(report);
        for (final NotificationEventWithMetadata<AnalyticsReportJob> notification : getFutureNotificationsForReportJob(eventJson, transmogrifier)) {
            jobQueue.removeNotificationFromTransaction(transmogrifier, notification.getRecordId());
        }
    }

    public List<AnalyticsReportJob> schedules() {
        final List<AnalyticsReportJob> schedules = new LinkedList<AnalyticsReportJob>();
        for (final NotificationEventWithMetadata<AnalyticsReportJob> notification : getFutureNotifications(null)) {
            schedules.add(notification.getEvent());
        }
        return schedules;
    }

    private List<NotificationEventWithMetadata<AnalyticsReportJob>> getFutureNotifications(@Nullable Transmogrifier transmogrifier) {
        if (transmogrifier == null) {
            return jobQueue.getFutureNotificationForSearchKey2(AnalyticsReportJob.class, JOBS_SCHEDULER_VERSION);
        } else {
            return jobQueue.getFutureNotificationFromTransactionForSearchKey2(AnalyticsReportJob.class, JOBS_SCHEDULER_VERSION, transmogrifier);
        }
    }

    private Iterable<NotificationEventWithMetadata<AnalyticsReportJob>> getFutureNotificationsForReportJob(final AnalyticsReportJob reportJob, @Nullable Transmogrifier transmogrifier) {
        final Integer eventJsonRecordId = reportJob.getRecordId();
        if (eventJsonRecordId != null) {
            // Fast search path<commit_msg>scheduler: return schedules in a deterministic order<commit_after>/*
 * Copyright 2010-2014 Ning, Inc.
 * Copyright 2014 The Billing Project, LLC
 *
 * Ning licenses this file to you under the Apache License, version 2.0
 * (the "License"); you may not use this file except in compliance with the
 * License.  You may obtain a copy of the License at:
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */

package org.killbill.billing.plugin.analytics.reports.scheduler;

import java.io.IOException;
import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.UUID;

import javax.annotation.Nullable;

import com.google.common.collect.Ordering;
import org.joda.time.DateTime;
import org.killbill.billing.plugin.analytics.dao.BusinessDBIProvider;
import org.killbill.billing.plugin.analytics.reports.configuration.ReportsConfigurationModelDao;
import org.killbill.billing.plugin.analytics.reports.configuration.ReportsConfigurationModelDao.Frequency;
import org.killbill.clock.Clock;
import org.killbill.killbill.osgi.libs.killbill.OSGIKillbillDataSource;
import org.killbill.killbill.osgi.libs.killbill.OSGIKillbillLogService;
import org.killbill.notificationq.DefaultNotificationQueueService;
import org.killbill.notificationq.api.NotificationEvent;
import org.killbill.notificationq.api.NotificationEventWithMetadata;
import org.killbill.notificationq.api.NotificationQueue;
import org.killbill.notificationq.api.NotificationQueueService.NotificationQueueAlreadyExists;
import org.killbill.notificationq.api.NotificationQueueService.NotificationQueueHandler;
import org.osgi.service.log.LogService;
import org.skife.jdbi.v2.Call;
import org.skife.jdbi.v2.Handle;
import org.skife.jdbi.v2.IDBI;
import org.skife.jdbi.v2.sqlobject.mixins.Transmogrifier;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Objects;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;

import static org.killbill.billing.plugin.analytics.AnalyticsActivator.ANALYTICS_QUEUE_SERVICE;

public class JobsScheduler {

    // Current version of the jobs in the notification queue
    // This is useful to retrieve all currently scheduled ones
    private static final Long JOBS_SCHEDULER_VERSION = 1L;

    private static final Ordering<AnalyticsReportJob> ANALYTICS_REPORT_JOB_ORDERING = Ordering.from(new Comparator<AnalyticsReportJob>() {
        @Override
        public int compare(AnalyticsReportJob o1, AnalyticsReportJob o2) {
            return o1.getRecordId().compareTo(o2.getRecordId());
        }
    });

    private final OSGIKillbillLogService logService;
    private final IDBI dbi;
    private final Clock clock;
    private final NotificationQueue jobQueue;

    public JobsScheduler(final OSGIKillbillLogService logService,
                         final OSGIKillbillDataSource osgiKillbillDataSource,
                         final Clock clock,
                         final DefaultNotificationQueueService notificationQueueService) throws NotificationQueueAlreadyExists {
        this.logService = logService;
        this.clock = clock;

        dbi = BusinessDBIProvider.get(osgiKillbillDataSource.getDataSource());
        final NotificationQueueHandler notificationQueueHandler = new NotificationQueueHandler() {

            @Override
            public void handleReadyNotification(final NotificationEvent eventJson, final DateTime eventDateTime, final UUID userToken, final Long searchKey1, final Long searchKey2) {
                if (eventJson == null || !(eventJson instanceof AnalyticsReportJob)) {
                    logService.log(LogService.LOG_ERROR, "Analytics report service received an unexpected event: " + eventJson);
                    return;
                }

                final AnalyticsReportJob job = (AnalyticsReportJob) eventJson;
                logService.log(LogService.LOG_INFO, "Starting job for " + job.getReportName());

                try {
                    callStoredProcedure(job.getRefreshProcedureName());
                } finally {
                    schedule(job, null);
                    logService.log(LogService.LOG_INFO, "Ending job for " + job.getReportName());
                }
            }
        };
        jobQueue = notificationQueueService.createNotificationQueue(ANALYTICS_QUEUE_SERVICE,
                                                                    "reports-jobs",
                                                                    notificationQueueHandler);
    }

    public void start() {
        jobQueue.startQueue();
    }

    public void shutdownNow() {
        jobQueue.stopQueue();
    }

    public void scheduleNow(final ReportsConfigurationModelDao report) {
        final AnalyticsReportJob eventJson = new AnalyticsReportJob(report);
        schedule(eventJson, clock.getUTCNow(), null);
    }

    public void schedule(final ReportsConfigurationModelDao report, final Transmogrifier transmogrifier) {
        final AnalyticsReportJob eventJson = new AnalyticsReportJob(report);
        schedule(eventJson, transmogrifier);
    }

    public void unSchedule(final ReportsConfigurationModelDao report, final Transmogrifier transmogrifier) {
        final AnalyticsReportJob eventJson = new AnalyticsReportJob(report);
        for (final NotificationEventWithMetadata<AnalyticsReportJob> notification : getFutureNotificationsForReportJob(eventJson, transmogrifier)) {
            jobQueue.removeNotificationFromTransaction(transmogrifier, notification.getRecordId());
        }
    }

    public List<AnalyticsReportJob> schedules() {
        final List<AnalyticsReportJob> schedules = new LinkedList<AnalyticsReportJob>();
        for (final NotificationEventWithMetadata<AnalyticsReportJob> notification : getFutureNotifications(null)) {
            schedules.add(notification.getEvent());
        }
        return ANALYTICS_REPORT_JOB_ORDERING.immutableSortedCopy(schedules);
    }

    private List<NotificationEventWithMetadata<AnalyticsReportJob>> getFutureNotifications(@Nullable Transmogrifier transmogrifier) {
        if (transmogrifier == null) {
            return jobQueue.getFutureNotificationForSearchKey2(AnalyticsReportJob.class, JOBS_SCHEDULER_VERSION);
        } else {
            return jobQueue.getFutureNotificationFromTransactionForSearchKey2(AnalyticsReportJob.class, JOBS_SCHEDULER_VERSION, transmogrifier);
        }
    }

    private Iterable<NotificationEventWithMetadata<AnalyticsReportJob>> getFutureNotificationsForReportJob(final AnalyticsReportJob reportJob, @Nullable Transmogrifier transmogrifier) {
        final Integer eventJsonRecordId = reportJob.getRecordId();
        if (eventJsonRecordId != null) {
            // Fast search path
{'fraction_non_alphanumeric': 0.07265704707768547, 'fraction_numerical': 0.003935286401399213, 'mean_word_length': 3.48171129980405, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2566238', 'n_tokens_mistral': 3656, 'n_tokens_neox': 3420, 'n_words': 799}