public interface AggregatesAlertSender extends AlertSender {
    //only used for decoupling the AggregatesEmailAlarmCallback from the original EmailAlarmCallback and its binding to the AlertSender interface.
} public class AggregatesEmailAlarmCallback implements AlarmCallback {
    private static final Logger LOG = LoggerFactory.getLogger(AggregatesEmailAlarmCallback.class);

    public static final String CK_USER_RECEIVERS = "user_receivers";
    public static final String CK_EMAIL_RECEIVERS = "email_receivers";

    private final AggregatesAlertSender alertSender;
    private final NotificationService notificationService;
    private final NodeId nodeId;
    private final EmailRecipients.Factory emailRecipientsFactory;
    private final UserService userService;
    private final EmailConfiguration emailConfiguration;
    private Configuration configuration;
    private org.graylog2.Configuration graylogConfig;

    @Inject
    public AggregatesEmailAlarmCallback(AggregatesAlertSender alertSender,
                                        NotificationService notificationService,
                                        NodeId nodeId,
                                        EmailRecipients.Factory emailRecipientsFactory,
                                        UserService userService,
                                        EmailConfiguration emailConfiguration,
                                        org.graylog2.Configuration graylogConfig) {
        this.alertSender = alertSender;
        this.notificationService = notificationService;
        this.nodeId = nodeId;
        this.emailRecipientsFactory = emailRecipientsFactory;
        this.userService = userService;
        this.emailConfiguration = emailConfiguration;
        this.graylogConfig = graylogConfig;
    }

    @Override
    public void call(Stream stream, AlertCondition.CheckResult result) throws AlarmCallbackException {
        // Send alerts.
        final EmailRecipients emailRecipients = this.getEmailRecipients();
        if (emailRecipients.isEmpty()) {
            if (!emailConfiguration.isEnabled()) {
                throw new AlarmCallbackException("Email transport is not enabled in server configuration file!");
            }

            LOG.info("Alarm callback has no email recipients, not sending any emails.");
            return;
        }

        AlertCondition alertCondition = result.getTriggeredCondition();
        if (!(alertCondition instanceof AggregatesAlertCondition)) {
            throw new AlarmCallbackException("This Alarm Callback only works in conjunction with alerts generated by the Aggregates Plugin.");
        }

        try {
            if (alertCondition.getBacklog() > 0 && result.getMatchingMessages() != null) {
                alertSender.sendEmails(stream, emailRecipients, result, getAlarmBacklog(result));
            } else {
                alertSender.sendEmails(stream, emailRecipients, result);
            }
        } catch (TransportConfigurationException e) {
            LOG.warn("Alarm callback has email recipients and is triggered, but email transport is not configured.");
            Notification notification = notificationService.buildNow()
                    .addNode(nodeId.toString())
                    .addType(Notification.Type.EMAIL_TRANSPORT_CONFIGURATION_INVALID)
                    .addSeverity(Notification.Severity.NORMAL)
                    .addDetail("stream_id", stream.getId())
                    .addDetail("exception", e.getMessage());
            notificationService.publishIfFirst(notification);

            throw new AlarmCallbackException(e.getMessage(), e);
        } catch (Exception e) {
            LOG.error("Alarm callback has email recipients and is triggered, but sending emails failed", e);

            String exceptionDetail = e.toString();
            if (e.getCause() != null) {
                exceptionDetail += " (" + e.getCause() + ")";
            }

            Notification notification = notificationService.buildNow()
                    .addNode(nodeId.toString())
                    .addType(Notification.Type.EMAIL_TRANSPORT_FAILED)
                    .addSeverity(Notification.Severity.NORMAL)
                    .addDetail("stream_id", stream.getId())
                    .addDetail("exception", exceptionDetail);
            notificationService.publishIfFirst(notification);

            throw new AlarmCallbackException(e.getMessage(), e);
        }
    }

    private EmailRecipients getEmailRecipients() {

        return emailRecipientsFactory.create(
                configuration.getList(CK_USER_RECEIVERS, Collections.emptyList()),
                configuration.getList(CK_EMAIL_RECEIVERS, Collections.emptyList())
        );
    }

    protected List<Message> getAlarmBacklog(AlertCondition.CheckResult result) {
        final AlertCondition alertCondition = result.getTriggeredCondition();
        final List<MessageSummary> matchingMessages = result.getMatchingMessages();

        final int effectiveBacklogSize = Math.min(alertCondition.getBacklog(), matchingMessages.size());

        if (effectiveBacklogSize == 0) {
            return Collections.emptyList();
        }

        final List<MessageSummary> backlogSummaries = matchingMessages.subList(0, effectiveBacklogSize);

        final List<Message> backlog = Lists.newArrayListWithCapacity(effectiveBacklogSize);

        for (MessageSummary messageSummary : backlogSummaries) {
            backlog.add(messageSummary.getRawMessage());
        }

        return backlog;
    }

    @Override
    public void initialize(Configuration config) throws AlarmCallbackConfigurationException {
        this.configuration = config;
        this.alertSender.initialize(configuration);
    }

    // I am truly sorry about this, but leaking the user list is not okay...
    private ConfigurationRequest getConfigurationRequest(Map<String, String> userNames) {
        ConfigurationRequest configurationRequest = new ConfigurationRequest();
        configurationRequest.addField(new TextField("sender",
                "Sender",
                "graylog@example.org",
                "The sender of sent out mail alerts",
                ConfigurationField.Optional.OPTIONAL));

        configurationRequest.addField(new TextField("subject",
                "E-Mail Subject",
                "Graylog alert for stream: ${stream.title}: ${check_result.resultDescription}",
                "The subject of sent out mail alerts",
                ConfigurationField.Optional.NOT_OPTIONAL));

        configurationRequest.addField(new TextField("body",
                "E-Mail Body",
                FormattedEmailAlertSender.bodyTemplate,
                "The template to generate the body from",
                ConfigurationField.Optional.OPTIONAL,
                TextField.Attribute.TEXTAREA));

        configurationRequest.addField(new ListField(CK_EMAIL_RECEIVERS,
                "E-Mail Receivers",
                Collections.emptyList(),
                Collections.emptyMap(),
                "E-Mail addresses that should receive this alert",
                ConfigurationField.Optional.NOT_OPTIONAL,
                ListField.Attribute.ALLOW_CREATE));

        return configurationRequest;
    }

    @Override
    public ConfigurationRequest getRequestedConfiguration() {
        return getConfigurationRequest(Collections.emptyMap());
    }

    /* This method should be used when we want to provide user auto-completion to users that have permissions for it */
    public ConfigurationRequest getEnrichedRequestedConfiguration() {
        final Map<String, String> regularUsers = userService.loadAll().stream()
                .collect(Collectors.toMap(User::getName, User::getName));

        final Map<String, String> userNames = ImmutableMap.<String, String>builder()
                .put(graylogConfig.getRootUsername(), graylogConfig.getRootUsername())
                .putAll(regularUsers)
                .build();

        return getConfigurationRequest(userNames);
    }

    @Override
    public String getName() {
        return "Aggregates Email Alert Callback";
    }

    @Override
    public Map<String, Object> getAttributes() {
        return configuration.getSource();
    }

    @Override
    public void checkConfiguration() throws ConfigurationException {
        final boolean missingSender = isNullOrEmpty(configuration.getString("sender")) && isNullOrEmpty(emailConfiguration.getFromEmail());
        if (missingSender || isNullOrEmpty(configuration.getString("subject"))) {
            throw new ConfigurationException("Sender or subject are missing or invalid.");
        }
    }
} public class AggregatesAlertCondition extends AbstractAlertCondition {
    private static final Logger LOG = LoggerFactory.getLogger(AggregatesAlertCondition.class);
    private final String description;
    private final String query;
    private final String field;
    private final Long numberOfMatches;
    private final boolean matchMoreOrEqual;
    private final Searches searches;
    private final int limit;
    private final int interval;
    private final HistoryItemService historyItemService;
    private final String ruleName;
    private final ClusterConfigService clusterConfigService;

    enum CheckType {
        TERMS("terms count");

        private final String description;

        CheckType(String description) {
            this.description = description;
        }

        public String getDescription() {
            return description;
        }
    }

    public enum ThresholdType {
        LESS, MORE_OR_EQUAL
    }

    @AssistedInject
    public AggregatesAlertCondition(Searches searches,
                                    ClusterConfigService clusterConfigService,
                                    HistoryItemService historyItemService,
                                    @Assisted Stream stream,
                                    @Nullable @Assisted("id") String id,
                                    @Assisted DateTime createdAt,
                                    @Assisted("userid") String creatorUserId,
                                    @Assisted Map<String, Object> parameters,
                                    @Assisted("title") @Nullable String title) {
        super(stream, id, AggregatesUtil.ALERT_CONDITION_TYPE, createdAt, creatorUserId, parameters, title);

        this.description = (String) parameters.get("description");
        this.query = (String) parameters.get("query");
        this.field = (String) parameters.get("field");
        this.numberOfMatches = (Long)parameters.get("number_of_matches");
        this.matchMoreOrEqual = parameters.get("match_more_or_equal") == null ? true : (boolean) parameters.get("match_more_or_equal");
        this.searches = searches;
        this.limit = 100;
        this.interval = Tools.getNumber(parameters.get("interval"), Integer.valueOf(1)).intValue();
        this.ruleName = (String) parameters.get("rule_name");

        this.clusterConfigService = clusterConfigService;
        this.historyItemService= historyItemService;
    }

    public String getQuery() { return this.query; }
    public String getField() { return this.field; }

    @Override
    public String getDescription() {
        return this.description;
    }

    @Override
    public CheckResult runCheck() {

        Integer backlogSize = getBacklog();
        boolean backlogEnabled = false;
        int searchLimit = 100;

        if(backlogSize != null && backlogSize > 0) {
            backlogEnabled = true;
            searchLimit = backlogSize;
        }

        List<MessageSummary> summaries = Lists.newArrayListWithCapacity(searchLimit);

        String filter = "streams:" + stream.getId();

        final TimeRange timeRange = buildRelativeTimeRange(60 * this.interval);

        Map<String, Long> matchedTerms = new HashMap<String, Long>();
        TermsResult result = null;

        long ruleCount = 0;
        if (null != timeRange) {
            result = searches.terms(field, limit, query, filter, timeRange);

            LOG.debug("built query: " + result.getBuiltQuery());

            LOG.debug("query took " + result.tookMs() + "ms");

            for (Map.Entry<String, Long> term : result.getTerms().entrySet()) {

                String matchedFieldValue = term.getKey();
                Long count = term.getValue();

                if ((matchMoreOrEqual && count >= numberOfMatches)
                        || (!matchMoreOrEqual && count < numberOfMatches)) {

                    LOG.info(count + " found for " + field + "=" + matchedFieldValue);

                    matchedTerms.put(matchedFieldValue, count);
                    ruleCount += count;

                    if (backlogEnabled) {
                        SearchResult searchResult = searches.search(

                                query + " AND " + field + ": " + QueryParser.escape(matchedFieldValue),
                                filter,
                                timeRange,
                                searchLimit,
                                0,
                                new Sorting(Message.FIELD_TIMESTAMP, Sorting.Direction.DESC)
                        );




                        for (ResultMessage resultMessage : searchResult.getResults()) {
                            if (summaries.size() < searchLimit) {
                                final Message msg = resultMessage.getMessage();
                                summaries.add(new MessageSummary(resultMessage.getIndex(), msg));
                            } else {
                                break;
                            }
                        }
                    } else {

                        summaries = Collections.emptyList();
                        LOG.debug("No messages found");
                    }


                }

            }
        }

        if (result != null && (!matchedTerms.isEmpty() || (result.getTerms().size() == 0 && !matchMoreOrEqual))){
            HistoryItem historyItem = HistoryItemImpl.create(this.ruleName, new Date(), ruleCount);

            historyItemService.create(historyItem);

            LOG.debug("Alert check <{}> found [{}] terms.", id, matchedTerms.size());
            return new AggregatesCheckResult(true, this, this.description, Tools.nowUTC(), summaries, matchedTerms);
        } else {
            LOG.debug("Alert check <{}> found no terms, alert should be resolved.");
            return new NegativeCheckResult();
        }
    }


    public class AggregatesCheckResult extends AbstractAlertCondition.CheckResult {
        private Map<String, Long> matchedTerms;
        AggregatesCheckResult(boolean isTriggered, AlertCondition triggeredCondition, String resultDescription, DateTime triggeredAt, List<MessageSummary> summaries, Map<String, Long> matchedTerms) {
            super(isTriggered, triggeredCondition, resultDescription, triggeredAt, summaries);
            this.matchedTerms = matchedTerms;
        }

        public Map<String, Long> getMatchedTerms(){
            return matchedTerms;
        }
    }




    public static class Descriptor extends AlertCondition.Descriptor {
        public Descriptor() {

            super(
                    "Aggregate Rule Alert Condition",
                    "https://github.com/cvtienhoven/graylog-plugin-aggregates",
                    "This condition is triggered when an Aggregates Rule has been satisfied."

            );
        }
    }

    public static class Config implements AlertCondition.Config {
        public Config() {

        }

        @Override
        public ConfigurationRequest getRequestedConfiguration() {
            final ConfigurationRequest configurationRequest = ConfigurationRequest.createWithFields();
            configurationRequest.addFields(AbstractAlertCondition.getDefaultConfigurationFields());

            return configurationRequest;
        }
    }

    public interface Factory extends AlertCondition.Factory {
        @Override
        AggregatesAlertCondition create(Stream stream,
                                        @Assisted("id") String id,
                                        DateTime createdAt,
                                        @Assisted("userid") String creatorUserId,
                                        Map<String, Object> parameters,
                                        @Assisted("title") @Nullable String title);

        @Override
        AggregatesAlertCondition.Config config();

        @Override
        AggregatesAlertCondition.Descriptor descriptor();
    }

    @VisibleForTesting
    TimeRange buildRelativeTimeRange(int range) {
        try {
            return restrictTimeRange(RelativeRange.create(range));
        } catch (InvalidRangeParametersException e) {
            LOG.warn("Invalid timerange parameters provided, not executing rule");
            return null;
        }
    }

    protected org.graylog2.plugin.indexer.searches.timeranges.TimeRange restrictTimeRange(
            final org.graylog2.plugin.indexer.searches.timeranges.TimeRange timeRange) {
        final DateTime originalFrom = timeRange.getFrom();
        final DateTime to = timeRange.getTo();
        final DateTime from;

        final SearchesClusterConfig config = clusterConfigService.get(SearchesClusterConfig.class);

        if (config == null || Period.ZERO.equals(config.queryTimeRangeLimit())) {
            from = originalFrom;
        } else {
            final DateTime limitedFrom = to.minus(config.queryTimeRangeLimit());
            from = limitedFrom.isAfter(originalFrom) ? limitedFrom : originalFrom;
        }

        return AbsoluteRange.create(from, to);
    }

    public boolean parametersEqual(Map<String, Object> parameters){
        if (this.description == null || !this.description.equals((String) parameters.get("description"))){
            return false;
        }
        if (this.query == null || !this.query.equals((String) parameters.get("query"))){
            return false;
        }
        if (this.ruleName == null || !this.ruleName.equals((String) parameters.get("rule_name"))){
            return false;
        }
        if (this.field == null || !this.field.equals((String) parameters.get("field"))){
            return false;
        }
        if (!this.numberOfMatches.equals((Long)parameters.get("number_of_matches"))){
            return false;
        }
        if (this.matchMoreOrEqual != (parameters.get("match_more_or_equal") == null ? true : (boolean) parameters.get("match_more_or_equal"))){
            return false;
        }
        if (this.repeatNotifications != (boolean) parameters.get("repeat_notifications")){
            return false;
        }
        if (this.interval != Tools.getNumber(parameters.get("interval"), Integer.valueOf(1)).intValue()){
            return false;
        }
        return true;
    }

} public class FormattedEmailAlertSender implements AggregatesAlertSender {
    private static final Logger LOG = LoggerFactory.getLogger(org.graylog2.alerts.FormattedEmailAlertSender.class);
    private static final String FONT_SIZE = "14px";
    private static final String FONT_FAMILY = "Arial";

    public static final String bodyTemplate = "##########\n" +
            "Alert Description: ${check_result.resultDescription}\n" +
            "Date: ${check_result.triggeredAt}\n" +
            "Stream ID: ${stream.id}\n" +
            "Stream title: ${stream.title}\n" +
            "Stream description: ${stream.description}\n" +
            "Alert Condition Title: ${alertCondition.title}\n" +
            "${if stream_url}Stream URL: ${stream_url}${end}\n" +
            "\n" +
            "${matchedTermsTable}\n" +
            "\n" +
            "Triggered condition: ${check_result.triggeredCondition}\n" +
            "##########\n\n" +
            "${if backlog}" +
            "Last messages accounting for this alert:\n" +
            "${foreach backlog message}" +
            "${message}\n\n" +
            "${end}" +
            "${else}" +
            "<No backlog>\n" +
            "${end}" +
            "\n";

    private final Engine templateEngine;
    private final NotificationService notificationService;
    private final NodeId nodeId;
    private Configuration pluginConfig;

    private final EmailConfiguration configuration;

    @Inject
    public FormattedEmailAlertSender(EmailConfiguration configuration,
                                     NotificationService notificationService,
                                     NodeId nodeId,
                                     Engine templateEngine) {
        this.configuration = requireNonNull(configuration, "configuration");
        this.notificationService = requireNonNull(notificationService, "notificationService");
        this.nodeId = requireNonNull(nodeId, "nodeId");
        this.templateEngine = requireNonNull(templateEngine, "templateEngine");
    }

    @Override
    public void initialize(Configuration configuration) {
        this.pluginConfig = configuration;
    }

    @VisibleForTesting
    String buildSubject(Stream stream, AlertCondition.CheckResult checkResult, List<Message> backlog) {
        final String template;
        if (pluginConfig == null || pluginConfig.getString("subject") == null) {
            template = "Graylog alert for stream: ${stream.title}: ${check_result.resultDescription}";
        } else {
            template = pluginConfig.getString("subject");
        }

        Map<String, Object> model = getModel(stream, checkResult, backlog);

        return templateEngine.transform(template, model).replaceAll("(\r\n|\n)", "<br />");
    }

    @VisibleForTesting
    String buildBody(Stream stream, AlertCondition.CheckResult checkResult, List<Message> backlog) {
        final String template;
        if (pluginConfig == null || pluginConfig.getString("body") == null) {
            template = bodyTemplate;
        } else {
            template = pluginConfig.getString("body");
        }
        Map<String, Object> model = getModel(stream, checkResult, backlog);

        return "<html><body style=\"font-family:" + FONT_FAMILY + "; font-size: " + FONT_SIZE + "\">" + this.templateEngine.transform(template, model) + "</body></html>";
    }

    private Map<String, Object> getModel(Stream stream, AlertCondition.CheckResult checkResult, List<Message> backlog) {
        String baseUri = buildStreamDetailsURL(configuration.getWebInterfaceUri(), checkResult, stream);
        Map<String, Object> model = new HashMap<>();
        model.put("stream", stream);
        model.put("check_result", checkResult);
        model.put("stream_url", buildStreamDetailsURL(configuration.getWebInterfaceUri(), checkResult, stream));
        model.put("alertCondition", checkResult.getTriggeredCondition());

        final List<Message> messages = firstNonNull(backlog, Collections.<Message>emptyList());
        model.put("backlog", messages);
        model.put("backlog_size", messages.size());
        model.put("matchedTermsTable", getMatchedTermsHTMLTable(checkResult, baseUri));
        return model;
    }

    private String getMatchedTermsHTMLTable(AlertCondition.CheckResult result, String baseUri){
        String field = ((AggregatesAlertCondition)result.getTriggeredCondition()).getField();
        String table = "<table width=\"100%\" border=\"1\" style=\"font-family:Arial; border-collapse: collapse; text-align: left;font-size: " + FONT_SIZE + ";\">";
        table += "<tr><th align=\"left\">Value of field \"" + field + "\"</th><th align=\"left\">Occurrences</th></tr>\n";
        for (Map.Entry<String, Long> entry : ((AggregatesAlertCondition.AggregatesCheckResult)result).getMatchedTerms().entrySet())
        {
            try {
                table += "<tr><td>" + entry.getKey() + "</td><td><a href=\"" + baseUri + URLEncoder.encode(" AND " + field + ": \"" + entry.getKey() + "\"", "UTF-8") +"\">"+entry.getValue() + "</a></td></tr>\n";
            } catch (UnsupportedEncodingException e) {
                LOG.error("Could not encode the <field=value> string.");
            }
        }
        table += "</table>";

        return table;
    }


    private String buildStreamDetailsURL(URI baseUri, AlertCondition.CheckResult checkResult, Stream stream) {
        // Return an informational message if the web interface URL hasn't been set
        if (baseUri == null || isNullOrEmpty(baseUri.getHost())) {
            return "Please configure 'transport_email_web_interface_url' in your Graylog configuration file.";
        }

        int time = 5;
        if (checkResult.getTriggeredCondition().getParameters().get("time") != null) {
            time = (int) checkResult.getTriggeredCondition().getParameters().get("time");
        }

        DateTime dateAlertEnd = checkResult.getTriggeredAt();
        DateTime dateAlertStart = dateAlertEnd.minusMinutes(time);
        String alertStart = Tools.getISO8601String(dateAlertStart);
        String alertEnd = Tools.getISO8601String(dateAlertEnd);

        AggregatesAlertCondition condition = (AggregatesAlertCondition) checkResult.getTriggeredCondition();

        String query = condition.getQuery();
        if (query != null && !"".equals(query)){
            try {
                query= "&q=" + URLEncoder.encode(query,"UTF-8");
            } catch (UnsupportedEncodingException e) {
                LOG.error("Failed to encode query [{}]", query );
            }
        } else {
            query = "";
        }

        return baseUri + "/streams/" + stream.getId() + "/messages?rangetype=absolute&from=" + alertStart + "&to=" + alertEnd + query;
    }

    @Override
    public void sendEmails(Stream stream, EmailRecipients recipients, AlertCondition.CheckResult checkResult) throws TransportConfigurationException, EmailException {
        sendEmails(stream, recipients, checkResult, null);
    }

    private void sendEmail(String emailAddress, Stream stream, AlertCondition.CheckResult checkResult, List<Message> backlog) throws TransportConfigurationException, EmailException {
        LOG.debug("Sending mail to " + emailAddress);
        if(!configuration.isEnabled()) {
            throw new TransportConfigurationException("Email transport is not enabled in server configuration file!");
        }

        final Email email = new HtmlEmail();
        email.setCharset(EmailConstants.UTF_8);

        if (isNullOrEmpty(configuration.getHostname())) {
            throw new TransportConfigurationException("No hostname configured for email transport while trying to send alert email!");
        } else {
            email.setHostName(configuration.getHostname());
        }
        email.setSmtpPort(configuration.getPort());
        if (configuration.isUseSsl()) {
            email.setSslSmtpPort(Integer.toString(configuration.getPort()));
        }

        if(configuration.isUseAuth()) {
            email.setAuthenticator(new DefaultAuthenticator(
                Strings.nullToEmpty(configuration.getUsername()),
                Strings.nullToEmpty(configuration.getPassword())
            ));
        }

        email.setSSLOnConnect(configuration.isUseSsl());
        email.setStartTLSEnabled(configuration.isUseTls());
        if (pluginConfig != null && !isNullOrEmpty(pluginConfig.getString("sender"))) {
            email.setFrom(pluginConfig.getString("sender"));
        } else {
            email.setFrom(configuration.getFromEmail());
        }

        email.setSubject(buildSubject(stream, checkResult, backlog));
        email.setMsg(buildBody(stream, checkResult, backlog));
        email.addTo(emailAddress);

        LOG.debug("Sending email to [{}]", emailAddress);

        email.send();
    }

    @Override
    public void sendEmails(Stream stream, EmailRecipients recipients, AlertCondition.CheckResult checkResult, List<Message> backlog) throws TransportConfigurationException, EmailException {
        if(!configuration.isEnabled()) {
            throw new TransportConfigurationException("Email transport is not enabled in server configuration file!");
        }

        if (recipients == null || recipients.isEmpty()) {
            throw new RuntimeException("Cannot send emails: empty recipient list.");
        }

        final Set<String> recipientsSet = recipients.getEmailRecipients();
        if (recipientsSet.size() == 0) {
            final Notification notification = notificationService.buildNow()
                .addNode(nodeId.toString())
                .addType(Notification.Type.GENERIC)
                .addSeverity(Notification.Severity.NORMAL)
                .addDetail("title", "Stream \"" + stream.getTitle() + "\" is alerted, but no recipients have been defined!")
                .addDetail("description", "To fix this, go to the alerting configuration of the stream and add at least one alert recipient.");
            notificationService.publishIfFirst(notification);
        }

        for (String email : recipientsSet) {
            sendEmail(email, stream, checkResult, backlog);
        }
    }
} public interface HistoryItemService {
    long count();

    HistoryItem create(HistoryItem historyItem);
    
    List<HistoryItem> all();

	List<HistoryAggregateItem> getForRuleName(String ruleName, int days);
	
	void removeBefore(Date date);

	List<HistoryAggregateItem> getForRuleName(String ruleName, String timespan);

    void updateHistoryRuleName(String oldName, String newName);
} public class HistoryItemServiceImpl implements HistoryItemService {
	private final static int SECONDS_IN_YEAR = 3600*24*366;
	private final static int SECONDS_IN_MONTH = 3600*24*31;
	private final static int SECONDS_IN_DAY = 3600*24;
	
	private final JacksonDBCollection<HistoryItemImpl, String> coll;
	private final Validator validator;
	private static final Logger LOG = LoggerFactory.getLogger(HistoryItemServiceImpl.class);

	@Inject
	public HistoryItemServiceImpl(MongoConnection mongoConnection, MongoJackObjectMapperProvider mapperProvider,
			Validator validator) {
		this.validator = validator;
		final String collectionName = HistoryItemImpl.class.getAnnotation(CollectionName.class).value();
		final DBCollection dbCollection = mongoConnection.getDatabase().getCollection(collectionName);
		this.coll = JacksonDBCollection.wrap(dbCollection, HistoryItemImpl.class, String.class, mapperProvider.get());
		// this.coll.createIndex(new BasicDBObject("name", 1), new
		// BasicDBObject("unique", true));
	}

	@Override
	public long count() {
		return coll.count();
	}

	@Override
	public HistoryItem create(HistoryItem historyItem) {
		if (historyItem instanceof HistoryItemImpl) {
			final HistoryItemImpl ruleImpl = (HistoryItemImpl) historyItem;
			final Set<ConstraintViolation<HistoryItemImpl>> violations = validator.validate(ruleImpl);
			if (violations.isEmpty()) {
				return coll.insert(ruleImpl).getSavedObject();

			} else {
				throw new IllegalArgumentException("Specified object failed validation: " + violations);
			}
		} else
			throw new IllegalArgumentException(
					"Specified object is not of correct implementation type (" + historyItem.getClass() + ")!");
	}

	@Override
	public List<HistoryItem> all() {
		return toAbstractListType(coll.find());
	}
	
	
	@SuppressWarnings("unchecked")
	@Override
	public List<HistoryAggregateItem> getForRuleName(String ruleName, String timespan) {
		Period period = Period.parse(timespan);
		Duration duration = period.toDurationFrom(new DateTime());		
		int seconds = duration.toStandardSeconds().getSeconds();
		
		Calendar c = Calendar.getInstance();
		c.add(Calendar.SECOND, seconds * -1);
		DBObject project = null;
		BasicDBObject additionalOperation = null;
		
		if (seconds <= SECONDS_IN_DAY) {
			project = (DBObject) JSON.parse("{'$project': { 'datePartHour' : {'$concat' : [{'$substr' : [{'$year' : '$timestamp'}, 0, 4]}, '-', {'$substr' : [{'$month' : '$timestamp'}, 0, 2]}, '-', {'$substr' : [{'$dayOfMonth' : '$timestamp'}, 0, 2]}, 'T',  {'$substr' : [{'$hour' : '$timestamp'}, 0, 2]}] }, 'numberOfHits':'$numberOfHits'}}");
			additionalOperation = new BasicDBObject("$group", (new BasicDBObject("_id",
					"$datePartHour").append("moment", new BasicDBObject("$first","$datePartHour"))).append("numberOfHits", new BasicDBObject("$sum", "$numberOfHits")));
		} else if (seconds <= SECONDS_IN_MONTH){
			project = (DBObject) JSON.parse("{'$project': { 'datePartDay' : {'$concat' : [ {'$substr' : [{'$year' : '$timestamp'}, 0, 4]}, '-', {'$substr' : [{'$month' : '$timestamp'}, 0, 2]}, '-', {'$substr' : [{'$dayOfMonth' : '$timestamp'}, 0, 2]}] }, 'numberOfHits':'$numberOfHits'}}");
			additionalOperation = new BasicDBObject("$group", (new BasicDBObject("_id",
					"$datePartDay").append("moment", new BasicDBObject("$first","$datePartDay"))).append("numberOfHits", new BasicDBObject("$sum", "$numberOfHits")));
		} else if (seconds <= SECONDS_IN_YEAR) {

			project = (DBObject) JSON.parse(
					"{'$project': { 'datePartMonth' : {'$concat' : [ {'$substr' : [{'$year' : '$timestamp'}, 0, 4]}, '-', {'$substr' : [{'$month' : '$timestamp'}, 0, 2]}] }, 'numberOfHits':'$numberOfHits'}}");
			additionalOperation = new BasicDBObject("$group", (new BasicDBObject("_id",
					"$datePartMonth").append("moment", new BasicDBObject("$first","$datePartMonth"))).append("numberOfHits", new BasicDBObject("$sum", "$numberOfHits")));
		} else {

			project = (DBObject) JSON.parse(
					"{'$project': { 'datePartYear' : {'$concat' : [ {'$substr' : [{'$year' : '$timestamp'}, 0, 4]}] }, 'numberOfHits':'$numberOfHits'}}");
			additionalOperation = new BasicDBObject("$group", (new BasicDBObject("_id",
					"$datePartYear").append("moment", new BasicDBObject("$first","$datePartYear"))).append("numberOfHits", new BasicDBObject("$sum", "$numberOfHits")));
		}
		
		
		
		DateFormat df = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
		
		//first match the records that meet the rule and the number of days 
		DBObject match = (DBObject) JSON.parse("{ '$match' : { $and: [ {'ruleName': '" + ruleName +"'}, {'timestamp': {'$gt' : { '$date': '" + df.format(c.getTime()) + "'}}}]}}");

		Aggregation<? extends HistoryAggregateItem> aggregation = new Aggregation<HistoryAggregateItemImpl>(HistoryAggregateItemImpl.class, match, project,
				additionalOperation);

		
		AggregationResult<? extends HistoryAggregateItem> aggregationResult = coll.aggregate(aggregation);

		LOG.debug("Aggregation result: " + aggregationResult.results().toString());
		
		return (List<HistoryAggregateItem>) aggregationResult.results();
				
	}

	@Override
	public void updateHistoryRuleName(String oldName, String newName) {
		WriteResult result = coll.updateMulti(DBQuery.is("ruleName", oldName), DBUpdate.set("ruleName", newName));
		LOG.info("Updated [{}] history items after rule rename from [{}] to [{}]", result.getN(), oldName, newName);
	}

	private List<HistoryItem> toAbstractListType(DBCursor<HistoryItemImpl> historyItems) {
		return toAbstractListType(historyItems.toArray());
	}

	private List<HistoryItem> toAbstractListType(List<HistoryItemImpl> historyItems) {
		final List<HistoryItem> result = Lists.newArrayListWithCapacity(historyItems.size());
		result.addAll(historyItems);
		LOG.debug("Number of history items returned: " + result.size());
		return result;
	}

	@Override
	public void removeBefore(Date date) {
		coll.remove(new BasicDBObject("timestamp", new BasicDBObject("$lt", date)));
		
	}

	@Override
	public List<HistoryAggregateItem> getForRuleName(String ruleName, int days) {
		// TODO Auto-generated method stub
		return null;
	}
} public class AggregatesMaintenance extends Periodical {
    private static final Logger LOG = LoggerFactory.getLogger(AggregatesMaintenance.class);
    private final AlertService alertService;
    private final StreamService streamService;
    private final RuleService ruleService;
    private final HistoryItemService historyItemService;
    private final ReportScheduleService reportScheduleService;
    private final ClusterConfigService clusterConfigService;
    private final ScheduledExecutorService scheduler;

    private final AtomicReference<AggregatesConfig> config;

    private static final int DEFAULT_RETENTION = 31 * 24 * 3600; // a month

    @Inject
    public AggregatesMaintenance(AlertService alertService, RuleService ruleService, StreamService streamService, HistoryItemService historyItemService, ReportScheduleService reportScheduleService, @Named("daemonScheduler") ScheduledExecutorService scheduler, ClusterConfigService clusterConfigService, EventBus eventBus) {
        this.alertService = alertService;
        this.ruleService = ruleService;
        this.streamService = streamService;
        this.historyItemService = historyItemService;
        this.reportScheduleService = reportScheduleService;
        this.clusterConfigService = clusterConfigService;
        this.scheduler = scheduler;

        final AggregatesConfig config = clusterConfigService.getOrDefault(AggregatesConfig.class,
                AggregatesConfig.defaultConfig());

        this.config = new AtomicReference<>(config);

        eventBus.register(this);
    }

    @Override
    public void doRun() {
        LOG.debug("Config: [{}]", config.get());

        if (config.get().purgeHistory()){
            purgeHistory();
        }

        if (config.get().resolveOrphanedAlerts()) {
            resolveOrphanedAlerts();
        }

    }


    private void purgeHistory(){
        Calendar cal = Calendar.getInstance();

        List<ReportSchedule> reportSchedules = reportScheduleService.all();
        int retention = AggregatesUtil.timespanToSeconds(config.get().historyRetention(), cal);

        for (ReportSchedule reportSchedule : reportSchedules) {
            int timespan = AggregatesUtil.timespanToSeconds(reportSchedule.getTimespan(), cal);
            if (timespan > retention) {
                retention = timespan;
            }
        }

        LOG.debug("Retention is set to " + retention + " seconds (" + new Duration(retention) + ")");

        cal.add(Calendar.SECOND, -1 * retention);

        //remove all items before the current date - retention time
        long initialCount = historyItemService.count();
        historyItemService.removeBefore(cal.getTime());
        LOG.debug("Removed " + (initialCount - historyItemService.count()) + " history items");

    }

    private void resolveOrphanedAlerts() {
        List<Rule> rules = ruleService.all();

        List<String> streamIds = new ArrayList<String>();
        for (Rule rule : rules) {
            if (rule.getAlertConditionId() != null && !streamIds.contains(rule.getStreamId())) {
                streamIds.add(rule.getStreamId());
            }

            if (rule.getAlertConditionId() == null){
                LOG.warn("Rule [{}] has no associated AlertCondition, perhaps it was created with an old version of the plugin. Please re-create it to be able to generate alerts.", rule.getName());
            }
        }

        LOG.info("Removing Aggregate Alert Conditions that don't have associated rule.");
        for (String streamId : streamIds) {
            LOG.debug("Checking stream [{}]", streamId);
            try {
                Stream triggeredStream = streamService.load(streamId);

                List<AlertCondition> alertConditions = streamService.getAlertConditions(triggeredStream);
                for (AlertCondition alertCondition : alertConditions) {

                    LOG.debug("AlertCondition: [{}]", alertCondition.getTitle());

                    resolveDuplicateAlerts(triggeredStream, alertCondition);

                    LOG.debug("Checking alert condition [{}] with type [{}]", alertCondition.getId(), alertCondition.getType());
                    if (alertCondition.getType().equals(AggregatesUtil.ALERT_CONDITION_TYPE)) {
                        LOG.debug("Type matches [{}], verifying if there's a rule with this alert condition", AggregatesUtil.ALERT_CONDITION_TYPE);
                        boolean found = false;
                        Rule foundRule = null;
                        for (Rule rule : rules) {
                            LOG.debug("Rule alertConditionId: [{}], alertConditionId: [{}]", rule.getAlertConditionId(), alertCondition.getId());
                            if (alertCondition.getId().equals(rule.getAlertConditionId())) {
                                foundRule = rule;
                                break;
                            }
                        }
                        if (foundRule == null) {
                            LOG.warn("Removing alertCondition [{}] of type [{}] because it's orphaned.", alertCondition.getTitle(), alertCondition.getType());
                            streamService.removeAlertCondition(triggeredStream, alertCondition.getId());
                        } else {
                            LOG.debug("AlertCondition [{}] is bound to rule [{}].", alertCondition.getTitle(), foundRule.getName());
                        }
                    }
                }


            } catch (NotFoundException e) {
                LOG.error("Stream with ID [{}] not found, skipping.", streamId);
                continue;
            }

        }

        //id, state, skip, limit
        LOG.info("Resolving unresolved Aggregate Alerts that don't have associated rule.");
        List<Alert> alerts = alertService.listForStreamIds(streamIds, Alert.AlertState.UNRESOLVED, 0, 100);
        for (Alert alert : alerts) {

            LOG.debug("Alert found [{}] with alertCondition [{}]", alert.getDescription(), alert.getConditionId());
            String alertConditionId = alert.getConditionId();

            try {
                Stream triggeredStream = streamService.load(alert.getStreamId());
                if (streamService.getAlertCondition(triggeredStream, alertConditionId).getType().equals(AggregatesUtil.ALERT_CONDITION_TYPE)) {
                    Rule foundRule = null;
                    for (Rule rule : rules) {
                        if (alertConditionId.equals(rule.getAlertConditionId())) {
                            foundRule = rule;
                            break;
                        }
                    }
                    if (foundRule == null) {
                        LOG.warn("Alert [{}] not bound to any rule, resolving", alert.getDescription());
                        alertService.resolveAlert(alert);
                    } else {
                        LOG.debug("Alert [{}] bound to rule [{}]", alert.getDescription(), foundRule.getName());
                    }
                } else {
                    LOG.debug("Alert [{}] does not belong to Aggregates AlertCondition", alert.getDescription());
                }
            } catch (NotFoundException e) {
                LOG.warn("Alert Condition not found for alert [{}], resolving", alert.getDescription());
                alertService.resolveAlert(alert);
            }
        }


    }

    /**
     * Resolve alert that has an equivalent (same timestamp, stream, alert condition) that was resolved.
     * Is needed because the unresolved alert sometimes never shows up in alertService.getLastTriggeredAlert().
     * @TODO: Investigate why duplicate alerts occur.
     * @param stream the stream
     * @param alertCondition the alert condition
     */
    private void resolveDuplicateAlerts(Stream stream, AlertCondition alertCondition){
        LOG.debug("Resolving duplicate alerts");
        Optional<Alert> lastTriggeredAlert = alertService.getLastTriggeredAlert(stream.getId(), alertCondition.getId());
        List<Alert> unresolvedAlerts = alertService.listForStreamIds(ImmutableList.of(stream.getId()), Alert.AlertState.UNRESOLVED, 0, 1);

        if (lastTriggeredAlert.isPresent()){
            LOG.debug("Last Triggered Alert found: [{}].", lastTriggeredAlert);
        } else {
            LOG.debug("Last Triggered Alert Alert not found.");
        }

        if (unresolvedAlerts.size() != 0){
            LOG.debug("Unresolved alert found: [{}].", unresolvedAlerts.get(0));

            if (lastTriggeredAlert.isPresent() && lastTriggeredAlert.get().getResolvedAt() != null && lastTriggeredAlert.get().getTriggeredAt().equals(unresolvedAlerts.get(0).getTriggeredAt())){
                LOG.debug("Last triggered alert is resolved at [{}], triggered at timestamp: [{}]", lastTriggeredAlert.get().getResolvedAt(), lastTriggeredAlert.get().getTriggeredAt());
                LOG.debug("Unresolved alert is triggered at timestamp: [{}]", lastTriggeredAlert.get().getTriggeredAt());

                LOG.info("Forcing resolve of alert [{}]", unresolvedAlerts.get(0));
                alertService.resolveAlert(unresolvedAlerts.get(0));
            }
        } else {
            LOG.debug("No unresolved duplicate alert found.");
        }
    }


    @Override
    public int getInitialDelaySeconds() {
        return 0;
    }

    @Override
    protected Logger getLogger() {
        return LOG;
    }

    @Override
    public int getPeriodSeconds() {
        return 60;
    }

    @Override
    public boolean isDaemon() {
        return true;
    }

    @Override
    public boolean masterOnly() {

        return true;
    }

    @Override
    public boolean runsForever() {
        return false;
    }

    @Override
    public boolean startOnThisNode() {
        return true;
    }

    @Override
    public boolean stopOnGracefulShutdown() {
        return true;
    }

    @Subscribe
    @SuppressWarnings("unused")
    public void updateConfig(ClusterConfigChangedEvent event) {
        if (!AggregatesConfig.class.getCanonicalName().equals(event.type())) {
            return;
        }
        scheduler.schedule((Runnable) this::reload, 0, TimeUnit.SECONDS);
        LOG.info("AggregatesConfig updated");
    }

    private void reload() {
        final AggregatesConfig newConfig = clusterConfigService.getOrDefault(AggregatesConfig.class,
                AggregatesConfig.defaultConfig());

        LOG.info("Updating AggregatesConfig - {}", newConfig);
        config.set(newConfig);
        //filterEngine.set(new AggregatesConfig(newConfig, metricRegistry));
    }
} public class RuleRestPermissions implements PluginPermissions {
    public static final String AGGREGATE_RULES_READ = "aggregate_rules:read";
    public static final String AGGREGATE_RULES_CREATE = "aggregate_rules:create";
    public static final String AGGREGATE_RULES_UPDATE = "aggregate_rules:update";
    public static final String AGGREGATE_RULES_DELETE = "aggregate_rules:delete";

    private final ImmutableSet<Permission> permissions = ImmutableSet.of(
            create(AGGREGATE_RULES_READ, "Read aggregate rules"),
            create(AGGREGATE_RULES_CREATE, "Create aggregate rules"),
            create(AGGREGATE_RULES_UPDATE, "Update aggregate rules"),
            create(AGGREGATE_RULES_DELETE, "Delete aggregate rules")
    );

    @Override
    public Set<Permission> permissions() {
        return permissions;
    }

    @Override
    public Set<Permission> readerBasePermissions() {
        return Collections.emptySet();
    }
} public class ReportScheduleRestPermissions implements PluginPermissions {
    public static final String AGGREGATE_REPORT_SCHEDULES_READ = "aggregate_report_schedules:read";
    public static final String AGGREGATE_REPORT_SCHEDULES_CREATE = "aggregate_report_schedules:create";
    public static final String AGGREGATE_REPORT_SCHEDULES_UPDATE = "aggregate_report_schedules:update";
    public static final String AGGREGATE_REPORT_SCHEDULES_DELETE = "aggregate_report_schedules:delete";

    private final ImmutableSet<Permission> permissions = ImmutableSet.of(
            create(AGGREGATE_REPORT_SCHEDULES_READ, "Read aggregate report schedules"),
            create(AGGREGATE_REPORT_SCHEDULES_CREATE, "Create aggregate report schedules"),
            create(AGGREGATE_REPORT_SCHEDULES_UPDATE, "Update aggregate report schedules"),
            create(AGGREGATE_REPORT_SCHEDULES_DELETE, "Delete aggregate report schedules")
    );

    @Override
    public Set<Permission> permissions() {
        return permissions;
    }

    @Override
    public Set<Permission> readerBasePermissions() {
        return Collections.emptySet();
    }
} public class AggregatesReport extends Periodical {
	private static final Logger LOG = LoggerFactory.getLogger(AggregatesReport.class);

	private final ReportSender reportSender;
	private final HistoryItemService historyItemService;
	private final RuleService ruleService;
	private final ReportScheduleService reportScheduleService;
	private String hostname = "localhost";

	@Inject
	public AggregatesReport(ReportSender reportSender, HistoryItemService historyItemService, RuleService ruleService,
			ReportScheduleService reportScheduleService) {
		this.reportSender = reportSender;
		this.historyItemService = historyItemService;
		this.ruleService = ruleService;
		this.reportScheduleService = reportScheduleService;

		InetAddress addr;
		try {
			addr = InetAddress.getLocalHost();
			this.hostname = addr.getCanonicalHostName();
		} catch (UnknownHostException e) {
			e.printStackTrace();
		}

	}

	private void setNewFireTime(ReportSchedule reportSchedule, Calendar cal) {
		CronExpression c;
		try {
			c = new CronExpression(reportSchedule.getExpression());
			reportScheduleService.updateNextFireTime(reportSchedule.getId(), c.getNextValidTimeAfter(cal.getTime()));
		} catch (ParseException e) {
			LOG.error("Schedule " + reportSchedule.getName() + " has invalid Cron Expression "
					+ reportSchedule.getExpression());
		}
	}

	private ReportSchedule getMatchingSchedule(Rule rule, List<ReportSchedule> reportSchedules) {
		for (ReportSchedule reportSchedule : reportSchedules) {
			if (rule.getReportSchedules().contains(reportSchedule.getId())) {
				return reportSchedule;
			}
		}
		return null;
	}

	@Override
	public void doRun() {
		Calendar cal = Calendar.getInstance();

		List<ReportSchedule> reportSchedules = reportScheduleService.all();

		List<ReportSchedule> applicableReportSchedules = new ArrayList<ReportSchedule>();

		// get the schedules that apply to the current dateTime
		for (ReportSchedule reportSchedule : reportSchedules) {
			if (reportSchedule.getNextFireTime() == null) {
				setNewFireTime(reportSchedule, cal);
			}

			if (reportSchedule.getNextFireTime() != null && new Date(reportSchedule.getNextFireTime()).before(cal.getTime())) {
				applicableReportSchedules.add(reportSchedule);
				setNewFireTime(reportSchedule, cal);
			}
		}

		// select the rules that match the applicable schedules
		List<Rule> rulesList = ruleService.all();
		Map<String, Map<Rule, List<HistoryAggregateItem>>> receipientsSeries = new HashMap<String, Map<Rule, List<HistoryAggregateItem>>>();
		Map<Rule, ReportSchedule>  ruleScheduleMapping = new HashMap<Rule, ReportSchedule>();
		
		for (Rule rule : rulesList) {
			if (rule.isInReport()) {
				ReportSchedule matchingSchedule = getMatchingSchedule(rule, applicableReportSchedules);
				
				if (matchingSchedule != null) {
					if (matchingSchedule.getReportReceivers() != null && matchingSchedule.getReportReceivers().size() > 0) {
						ruleScheduleMapping.put(rule, matchingSchedule);
						LOG.info("Rule \"" + rule.getName() + "\" will be added to report");
										
						for (String receipient : matchingSchedule.getReportReceivers()) {
							if (!receipientsSeries.containsKey(receipient)) {
								receipientsSeries.put(receipient, new HashMap<Rule, List<HistoryAggregateItem>>());
							}
							receipientsSeries.get(receipient).put(rule,
									historyItemService.getForRuleName(rule.getName(), matchingSchedule.getTimespan()));
						}
					} else {
						LOG.warn("No receivers found in Report Schedule \"" + matchingSchedule.getName() + "\", not using this schedule.");
					}
				
				}

			}
		}

		ByteArrayOutputStream outputStream = null;
		try {
			for (Map.Entry<String, Map<Rule, List<HistoryAggregateItem>>> receipientSeries : receipientsSeries
					.entrySet()) {
				outputStream = new ByteArrayOutputStream();
				
				ReportFactory.createReport(receipientSeries.getValue(), ruleScheduleMapping, cal, outputStream, hostname);
				byte[] bytes = outputStream.toByteArray();

				reportSender.sendEmail(receipientSeries.getKey(), bytes);
			}

		} catch (ParseException e) {
			LOG.error("Failed to create report, " + e.getMessage());
			LOG.debug("Stacktrace: " + e.getStackTrace());
		} catch (TransportConfigurationException e) {

			e.printStackTrace();
		} catch (EmailException e) {
			e.printStackTrace();
		} catch (MessagingException e) {
			e.printStackTrace();
		} finally {
			if (null != outputStream) {
				try {
					outputStream.close();
					outputStream = null;
				} catch (Exception ex) {
				}
			}
		}		
	}

	@Override
	public int getInitialDelaySeconds() {
		// TODO Auto-generated method stub
		return 0;
	}

	@Override
	protected Logger getLogger() {
		return LOG;
	}

	@Override
	public int getPeriodSeconds() {
		return 60;
	}

	@Override
	public boolean isDaemon() {
		return true;
	}

	@Override
	public boolean masterOnly() {

		return true;
	}

	@Override
	public boolean runsForever() {
		return false;
	}

	@Override
	public boolean startOnThisNode() {
		return true;
	}

	@Override
	public boolean stopOnGracefulShutdown() {
		return true;
	}

} public interface RuleService {
    long count();

    Rule update(String name, Rule rule);

    Rule create(Rule rule);
    
    List<Rule> all();
    
    int destroy(String ruleName) throws MongoException, UnsupportedEncodingException;

	Rule fromRequest(AddRuleRequest request);
	
	Rule fromRequest(UpdateRuleRequest request);

	//Rule setAlertConditionId(Rule rule, String currentAlertId);

    String createOrUpdateAlertCondition(Rule rule);
} public class RuleServiceImpl implements RuleService {

	private final JacksonDBCollection<RuleImpl, String> coll;
	private final Validator validator;
	private final StreamService streamService;
	private static final Logger LOG = LoggerFactory.getLogger(RuleServiceImpl.class);
	private final AlertConditionFactory alertConditionFactory;
	private final HistoryItemService historyItemService;

	@Inject
	public RuleServiceImpl(MongoConnection mongoConnection, MongoJackObjectMapperProvider mapperProvider,
                           Validator validator, StreamService streamService, AlertConditionFactory alertConditionFactory,
                           HistoryItemService historyItemService) {
		this.validator = validator;
		final String collectionName = RuleImpl.class.getAnnotation(CollectionName.class).value();
		final DBCollection dbCollection = mongoConnection.getDatabase().getCollection(collectionName);
		this.coll = JacksonDBCollection.wrap(dbCollection, RuleImpl.class, String.class, mapperProvider.get());
		this.coll.createIndex(new BasicDBObject("name", 1), new BasicDBObject("unique", true));
		this.streamService = streamService;
		this.alertConditionFactory = alertConditionFactory;
		this.historyItemService = historyItemService;
	}

	@Override
	public long count() {
		return coll.count();
	}

	@Override
	public Rule create(Rule rule) {
		if (rule instanceof RuleImpl) {
			final RuleImpl ruleImpl = (RuleImpl) rule;

			final Set<ConstraintViolation<RuleImpl>> violations = validator.validate(ruleImpl);
			if (violations.isEmpty()) {
				return coll.insert(ruleImpl).getSavedObject();

				//return createOrUpdateAlertCondition(ruleNew);
			} else {
				throw new IllegalArgumentException("Specified object failed validation: " + violations);
			}
		} else
			throw new IllegalArgumentException(
					"Specified object is not of correct implementation type (" + rule.getClass() + ")!");
	}
	
	@Override
	public Rule update(String name, Rule rule) {
		
		if (rule instanceof RuleImpl) {
			RuleImpl originalRule = coll.findOne(DBQuery.is("name", name));

			if (!originalRule.getStreamId().equals(rule.getStreamId())){
				removeAlertCondition(originalRule);
				createOrUpdateAlertCondition(rule);
			}

			final RuleImpl ruleImpl = (RuleImpl) rule;
			LOG.debug("Rule to be updated [{}]", ruleImpl);

			if (rule.getAlertConditionId() != null) {
				Stream triggeredStream = null;
				try {
					triggeredStream = streamService.load(rule.getStreamId());
				} catch (NotFoundException e) {
					LOG.error("Stream with ID [{}] not found, skipping update of alert condition", rule.getStreamId(), rule.getName());
				}


				if (triggeredStream != null) {
					Map<String, Object> parameters = AggregatesUtil.parametersFromRule(rule);

					String title = AggregatesUtil.alertConditionTitleFromRule(rule);

					AggregatesAlertCondition alertCondition;
					try {
						alertCondition = (AggregatesAlertCondition) streamService.getAlertCondition(triggeredStream, rule.getAlertConditionId());

						if (!alertCondition.parametersEqual(parameters)) {
                            LOG.info("Parameters of Alert Condition changed for rule [{}], updating Alert Condition", name);
                            createOrUpdateAlertCondition(rule);
						}
					} catch (NotFoundException e) {
						LOG.warn("Alert Condition removed for rule [{}], re-instantiating", rule.getName());
                        createOrUpdateAlertCondition(rule);
					}
				}
			}

			final Set<ConstraintViolation<RuleImpl>> violations = validator.validate(ruleImpl);
			if (violations.isEmpty()) {

				Rule newRule = coll.findAndModify(DBQuery.is("name", name), new BasicDBObject(), new BasicDBObject(),
						false, ruleImpl, true, false);
                if (!name.equals(rule.getName())) {
                    historyItemService.updateHistoryRuleName(name, rule.getName());
                }
                return newRule;

			} else {
				throw new IllegalArgumentException("Specified object failed validation: " + violations);
			}



		} else
			throw new IllegalArgumentException(
					"Specified object is not of correct implementation type (" + rule.getClass() + ")!");
	}


	@Override
	public List<Rule> all() {		
		return toAbstractListType(coll.find());
	}

	@Override
	public Rule fromRequest(AddRuleRequest request) {
		String alertConditionId = createOrUpdateAlertCondition(request.getRule());
		LOG.info("Creating new rule [{}] with alertConditionId [{}]", request.getRule(), alertConditionId);
		Rule rule =  RuleImpl.create(
				request.getRule().getQuery(), 
				request.getRule().getField(),
				request.getRule().getNumberOfMatches(), 
				request.getRule().isMatchMoreOrEqual(),
				request.getRule().getInterval(), 
				request.getRule().getName(),
				request.getRule().isEnabled(),
				request.getRule().getStreamId(),
				request.getRule().isInReport(),
				request.getRule().getReportSchedules(),
				alertConditionId,
				request.getRule().shouldRepeatNotifications(),
				request.getRule().getBacklog());


        return rule;

	}

	@Override
	public Rule fromRequest(UpdateRuleRequest request) {
	    Rule rule = RuleImpl.create(
				request.getRule().getQuery(), 
				request.getRule().getField(),
				request.getRule().getNumberOfMatches(), 
				request.getRule().isMatchMoreOrEqual(),
				request.getRule().getInterval(), 
				request.getRule().getName(),
				request.getRule().isEnabled(),
				request.getRule().getStreamId(),
				request.getRule().isInReport(),
				request.getRule().getReportSchedules(),
				request.getRule().getAlertConditionId(),
				request.getRule().shouldRepeatNotifications(),
				request.getRule().getBacklog());

           return rule;
	}

	public String createOrUpdateAlertCondition(Rule rule){

        String query = rule.getQuery();
        String streamId = rule.getStreamId();

        Map<String, Object> parameters = AggregatesUtil.parametersFromRule(rule);

        String title = AggregatesUtil.alertConditionTitleFromRule(rule);

        Stream triggeredStream = null;
        try {
            triggeredStream = streamService.load(rule.getStreamId());
        } catch (NotFoundException e) {
            LOG.error("Stream with ID [{}] not found", rule.getStreamId());

        }

        AggregatesAlertCondition alertCondition;

        String alertConditionId = null;
        if (rule.getAlertConditionId() != null){
            alertConditionId = rule.getAlertConditionId();
        }

        try {
            alertCondition = (AggregatesAlertCondition) alertConditionFactory.createAlertCondition(AggregatesUtil.ALERT_CONDITION_TYPE, triggeredStream, alertConditionId, DateTime.now(), "admin", parameters, title);

            if (alertConditionId != null) {
            	streamService.removeAlertCondition(triggeredStream, alertConditionId);
                streamService.addAlertCondition(triggeredStream, alertCondition);
                LOG.debug("Updated Alert Condition [{}]", alertCondition.getId());
            } else {
                streamService.addAlertCondition(triggeredStream, alertCondition);
				LOG.debug("Created Alert Condition [{}]", alertCondition.getId());
            }
            return alertCondition.getId();
        } catch (ConfigurationException|ValidationException e) {
            LOG.error("Failed to save Alert Condition for rule [{}] ", rule.getName());
        }
		return null;

    }

	private void removeAlertCondition(Rule rule){
		Stream triggeredStream = null;
		try {
			triggeredStream = streamService.load(rule.getStreamId());
			streamService.removeAlertCondition(triggeredStream,rule.getAlertConditionId());
		} catch (NotFoundException e) {
			LOG.error("Stream with ID [{}] not found", rule.getStreamId());
		}
	}


	@Override
	public int destroy(String ruleName) {
	    Rule rule = coll.findOne(DBQuery.is("name", ruleName));
	    removeAlertCondition(rule);
		return coll.remove(DBQuery.is("name", ruleName)).getN();
	}

	private List<Rule> toAbstractListType(DBCursor<RuleImpl> rules) {
		return toAbstractListType(rules.toArray());
	}

	private List<Rule> toAbstractListType(List<RuleImpl> rules) {
		final List<Rule> result = Lists.newArrayListWithCapacity(rules.size());
		result.addAll(rules);
	
		
		return result;
	}
} @Api(value = "Aggregates/Rules", description = "Management of Aggregation rules.")
@Path("/rules")
@Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.APPLICATION_JSON)
public class RuleResource extends RestResource implements PluginRestResource {
    private static final Logger LOG = LoggerFactory.getLogger(RuleResource.class);
    private final RuleService ruleService;  
    
    @Inject
    public RuleResource(RuleService ruleService) {
        this.ruleService = ruleService;
    }

    @GET
    @Timed
    @ApiOperation(value = "Lists all existing rules")
    @RequiresAuthentication
    @RequiresPermissions(RuleRestPermissions.AGGREGATE_RULES_READ)
    public RulesList list() {
        final List<Rule> rules = ruleService.all();   
        return RulesList.create(rules);
    }
    
    @PUT
    @Timed    
    @ApiOperation(value = "Create a rule")
    @RequiresAuthentication
    @RequiresPermissions(RuleRestPermissions.AGGREGATE_RULES_CREATE)
    @AuditEvent(type = AuditEventTypes.AGGREGATES_RULE_CREATE)
    @ApiResponses(value = {
            @ApiResponse(code = 400, message = "The supplied request is not valid.")
    })
    public Response create(
                             @ApiParam(name = "JSON body", required = true) @Valid @NotNull AddRuleRequest request
                             ) {
        final Rule rule = ruleService.fromRequest(request);
        LOG.info("Creating rule with name [{}]", rule.getName());
        ruleService.create(rule);

        return Response.accepted().build();
    }

    @POST
    @Path("/{name}")
    @Timed
    @RequiresAuthentication
    @RequiresPermissions(RuleRestPermissions.AGGREGATE_RULES_UPDATE)
    @AuditEvent(type = AuditEventTypes.AGGREGATES_RULE_UPDATE)
    @ApiOperation(value = "Update a rule")
    @ApiResponses(value = {
            @ApiResponse(code = 400, message = "The supplied request is not valid.")
    })
    public Response update(@ApiParam(name = "name", required = true)
    					   @PathParam("name") String name,
                             @ApiParam(name = "JSON body", required = true) @Valid @NotNull UpdateRuleRequest request
                             ) throws UnsupportedEncodingException {
        final Rule rule = ruleService.fromRequest(request);

        //ruleService.update(java.net.URLDecoder.decode(name, "UTF-8"), rule);
        ruleService.update(name, rule);
        return Response.accepted().build();
    }
    
    @DELETE
    @Path("/{name}")
    @RequiresAuthentication
    @RequiresPermissions(RuleRestPermissions.AGGREGATE_RULES_DELETE)
    @AuditEvent(type = AuditEventTypes.AGGREGATES_RULE_DELETE)
    @ApiOperation(value = "Delete a rule")
    @ApiResponses(value = {
            @ApiResponse(code = 404, message = "Rule not found."),
            @ApiResponse(code = 400, message = "Invalid ObjectId.")
    })
    public void delete(@ApiParam(name = "name", required = true)
                              @PathParam("name") String name
                              ) throws NotFoundException, MongoException, UnsupportedEncodingException, URISyntaxException {

        ruleService.destroy(name);
    }
} public interface ReportScheduleService {
    long count();

    ReportSchedule update(String name, ReportSchedule schedule);

    ReportSchedule create(ReportSchedule schedule);
    
    List<ReportSchedule> all();
    
    int destroy(String scheduleName) throws MongoException, UnsupportedEncodingException;

	ReportSchedule fromRequest(AddReportScheduleRequest request);
	
	ReportSchedule fromRequest(UpdateReportScheduleRequest request);
	
	ReportSchedule get(String id);
	
	ReportSchedule updateNextFireTime(String id, Date nextFireTime);

} public class ReportScheduleServiceImpl implements ReportScheduleService {

	private final JacksonDBCollection<ReportScheduleImpl, String> coll;
	private final JacksonDBCollection<RuleImpl, String> ruleColl;
	private final Validator validator;
	private static final Logger LOG = LoggerFactory.getLogger(ReportScheduleServiceImpl.class);

	@Inject
	public ReportScheduleServiceImpl(MongoConnection mongoConnection, MongoJackObjectMapperProvider mapperProvider,
			Validator validator) {
		
		this.validator = validator;

		final String collectionName = ReportScheduleImpl.class.getAnnotation(CollectionName.class).value();
		final DBCollection dbCollection = mongoConnection.getDatabase().getCollection(collectionName);
		this.coll = JacksonDBCollection.wrap(dbCollection, ReportScheduleImpl.class, String.class, mapperProvider.get());
		this.coll.createIndex(new BasicDBObject("name", 1), new BasicDBObject("unique", true));

		final String ruleCollectionName = RuleImpl.class.getAnnotation(CollectionName.class).value();
		final DBCollection ruleDbCollection = mongoConnection.getDatabase().getCollection(ruleCollectionName);
		this.ruleColl = JacksonDBCollection.wrap(ruleDbCollection, RuleImpl.class, String.class, mapperProvider.get());
		
		
		ReportSchedule defaultSchedule = null;
		
		try {
			defaultSchedule = coll.find(DBQuery.is("name", "Every Saturday, 23:59")).next();	
		} catch (Exception e) {
			LOG.debug("Default weekly ReportSchedule does not exist yet");
			defaultSchedule = coll.insert(ReportScheduleImpl.create(null, "Every Saturday, 23:59", "0 59 23 ? * SAT *", "P7D", true, null, null)).getSavedObject();
			LOG.debug("Created default weekly ReportSchedule with ID " + defaultSchedule.getId());
		}
				
		
		defaultSchedule = null;
		
		try {
			defaultSchedule = coll.find(DBQuery.is("name", "First day of month, 00:00")).next();	
		} catch (Exception e) {
			LOG.debug("Default monthly ReportSchedule does not exist yet");
			defaultSchedule = coll.insert(ReportScheduleImpl.create(null, "First day of month, 00:00", "0 0 0 1 1/1 ? *", "P1M", true, null, null)).getSavedObject();
			LOG.debug("Created default monthly ReportSchedule with ID " + defaultSchedule.getId());
		}

	}

	@Override
	public long count() {
		return coll.count();
	}

	@Override
	public ReportSchedule updateNextFireTime(String id, Date nextFireTime){
		LOG.debug("updateNextFireTime() - _id=" + id + ", date=" +nextFireTime);
		DBUpdate.Builder update = new DBUpdate.Builder();
		update.set("nextFireTime", nextFireTime.getTime());
		BasicDBObject query = new BasicDBObject();
	    query.put("_id", new ObjectId(id));
		return coll.findAndModify(query, update);
	}
	
	@Override
	public ReportSchedule create(ReportSchedule schedule) {
		if (schedule instanceof ReportScheduleImpl) {
			final ReportScheduleImpl scheduleImpl = (ReportScheduleImpl) schedule;
			final Set<ConstraintViolation<ReportScheduleImpl>> violations = validator.validate(scheduleImpl);
						
			if (violations.isEmpty()) {				
				try {				
					new CronExpression(scheduleImpl.getExpression());
				} catch (ParseException e) {
					LOG.error("Schedule " + scheduleImpl.getName() + " has invalid Cron Expression " + scheduleImpl.getExpression());
					throw new IllegalArgumentException("Schedule " + scheduleImpl.getName() + " has invalid Cron Expression " + scheduleImpl.getExpression());
				}
				
				ReportSchedule reportSchedule = coll.insert(scheduleImpl).getSavedObject();
				
				LOG.debug("created schedule with ID " + reportSchedule.getId());
				
				return reportSchedule;
			} else {
				throw new IllegalArgumentException("Specified object failed validation: " + violations);
			}
		} else
			throw new IllegalArgumentException(
					"Specified object is not of correct implementation type (" + schedule.getClass() + ")!");
	}
	
	@Override
	public ReportSchedule update(String name, ReportSchedule schedule) {		
		
		if (schedule instanceof ReportScheduleImpl) {
			final ReportScheduleImpl scheduleImpl = (ReportScheduleImpl) schedule;
			LOG.debug("updated schedule: " + scheduleImpl);
			final Set<ConstraintViolation<ReportScheduleImpl>> violations = validator.validate(scheduleImpl);
			if (violations.isEmpty()) {
				//return coll.update(DBQuery.is("name", name), ruleImpl, false, false).getSavedObject();
				return coll.findAndModify(DBQuery.is("name", name), new BasicDBObject(), new BasicDBObject(),
						false, scheduleImpl, true, false);
			} else {
				throw new IllegalArgumentException("Specified object failed validation: " + violations);
			}
		} else
			throw new IllegalArgumentException(
					"Specified object is not of correct implementation type (" + schedule.getClass() + ")!");
	}
	
	@Override
	public List<ReportSchedule> all() {		
		return toAbstractListType(coll.find());
	}

	@Override
	public ReportSchedule fromRequest(AddReportScheduleRequest request) {
		return ReportScheduleImpl.create(
				request.getReportSchedule().getId(),
				request.getReportSchedule().getName(), 
				request.getReportSchedule().getExpression(),
				request.getReportSchedule().getTimespan(),
				false,
				null,
				request.getReportSchedule().getReportReceivers());
	}

	@Override
	public ReportSchedule fromRequest(UpdateReportScheduleRequest request) {
		return ReportScheduleImpl.create(
				request.getReportSchedule().getId(),
				request.getReportSchedule().getName(), 
				request.getReportSchedule().getExpression(),
				request.getReportSchedule().getTimespan(),
				false,
				null,
				request.getReportSchedule().getReportReceivers());
	}
	
	@Override
	public int destroy(String id) {

		BasicDBObject query = new BasicDBObject();
	    query.put("reportSchedules", id);
		
		if (ruleColl.find(query).count() == 0){
			query = new BasicDBObject();
		    query.put("_id", new ObjectId(id));		    
			return coll.remove(query).getN();
		}
		throw new IllegalArgumentException(
				"There are still rules associated with schedule " + id );
		
	}

	private List<ReportSchedule> toAbstractListType(DBCursor<ReportScheduleImpl> rules) {
		return toAbstractListType(rules.toArray());
	}

	private List<ReportSchedule> toAbstractListType(List<ReportScheduleImpl> reportSchedules) {
		final List<ReportSchedule> result = Lists.newArrayListWithCapacity(reportSchedules.size());
		result.addAll(reportSchedules);

		return result;
	}

	@Override
	public ReportSchedule get(String id) {
		ReportSchedule reportSchedule = coll.find(DBQuery.is("_id", id)).next();
		return reportSchedule;
	}
} @Api(value = "Aggregates/Schedules", description = "Management of Aggregation rules.")
@Path("/schedules")
@Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.APPLICATION_JSON)
public class ReportScheduleResource extends RestResource implements PluginRestResource {
    private final ReportScheduleService reportScheduleService;
    
    @Inject
    public ReportScheduleResource(ReportScheduleService ruleService) {
        this.reportScheduleService = ruleService;
    }

    @GET
    @Timed
    @ApiOperation(value = "Lists all existing report schedules")
    @RequiresAuthentication
    @RequiresPermissions(ReportScheduleRestPermissions.AGGREGATE_REPORT_SCHEDULES_READ)
    public ReportSchedulesList list() {
        final List<ReportSchedule> reportSchedules = reportScheduleService.all();   
        return ReportSchedulesList.create(reportSchedules);
    }
    
    @PUT
    @Timed    
    @ApiOperation(value = "Create a report schedule")
    @RequiresAuthentication
    @RequiresPermissions(ReportScheduleRestPermissions.AGGREGATE_REPORT_SCHEDULES_CREATE)
    @AuditEvent(type = AuditEventTypes.AGGREGATES_REPORT_SCHEDULE_CREATE)
    @ApiResponses(value = {
            @ApiResponse(code = 400, message = "The supplied request is not valid.")
    })
    public Response create(
                             @ApiParam(name = "JSON body", required = true) @Valid @NotNull AddReportScheduleRequest request
                             ) {
        final ReportSchedule reportSchedule = reportScheduleService.fromRequest(request);

        reportScheduleService.create(reportSchedule);

        return Response.accepted().build();
    }

    @POST
    @Path("/{name}")
    @Timed
    @RequiresAuthentication
    @RequiresPermissions(ReportScheduleRestPermissions.AGGREGATE_REPORT_SCHEDULES_UPDATE)
    @AuditEvent(type = AuditEventTypes.AGGREGATES_REPORT_SCHEDULE_UPDATE)
    @ApiOperation(value = "Update a report schedule")
    @ApiResponses(value = {
            @ApiResponse(code = 400, message = "The supplied request is not valid.")
    })
    public Response update(@ApiParam(name = "name", required = true)
    					   @PathParam("name") String name,
                             @ApiParam(name = "JSON body", required = true) @Valid @NotNull UpdateReportScheduleRequest request
                             ) throws UnsupportedEncodingException {
        final ReportSchedule reportSchedule = reportScheduleService.fromRequest(request);

        reportScheduleService.update(java.net.URLDecoder.decode(name, "UTF-8"), reportSchedule);

        return Response.accepted().build();
    }
    
    @DELETE
    @Path("/{id}")
    @RequiresAuthentication
    @RequiresPermissions(ReportScheduleRestPermissions.AGGREGATE_REPORT_SCHEDULES_DELETE)
    @AuditEvent(type = AuditEventTypes.AGGREGATES_REPORT_SCHEDULE_DELETE)
    @ApiOperation(value = "Delete a report schedule")
    @ApiResponses(value = {
            @ApiResponse(code = 404, message = "Schedule not found."),
            @ApiResponse(code = 400, message = "Invalid ObjectId.")
    })
    public void delete(@ApiParam(name = "id", required = true)
                              @PathParam("id") String id
                              ) throws NotFoundException, MongoException, UnsupportedEncodingException {
        reportScheduleService.destroy(java.net.URLDecoder.decode(id, "UTF-8"));
    }
} public class AggregatesUtil {
    public final static String ALERT_CONDITION_TYPE = "Aggregates Alert";

	public static int timespanToSeconds(String timespan, Calendar cal){
		Period period = Period.parse(timespan);
		Duration duration = period.toDurationFrom(new DateTime(cal.getTime()));
		return duration.toStandardSeconds().getSeconds();
	}

	public static String getAlertConditionDescription(Rule rule){
		String matchDescriptor = rule.getNumberOfMatches() + " or more";
		if (!rule.isMatchMoreOrEqual()){
			matchDescriptor = "less than " + rule.getNumberOfMatches();
		}
		return "The same value of field '" + rule.getField() + "' occurs " + matchDescriptor + " times in a " + rule.getInterval() + " minute interval";
	}
	
	public String buildSummary(Rule rule, EmailConfiguration emailConfiguration, Map<String, Long> matchedTerms, TimeRange timeRange) throws UnsupportedEncodingException {

        final StringBuilder sb = new StringBuilder();

        sb.append("Matched values for field [ " + rule.getField() + " ]\n");

        for (Map.Entry<String, Long> entry : matchedTerms.entrySet()) {
        	
        	sb.append("\nValue: " + entry.getKey() + "\n");
            sb.append("Occurrences: " + entry.getValue() + "\n");                        

            if (!emailConfiguration.isEnabled()) {
                sb.append("\n");
            } else {
                String streamId = rule.getStreamId();
                String search_uri = "";

                if (streamId != null && streamId != "") {
                    search_uri += "/streams/" + streamId;
                }
                search_uri += "/search?rangetype=absolute&fields=message%2Csource%2C" + rule.getField() + "&from=" + timeRange.getFrom() + "&to=" + timeRange.getTo() + "&q=" + URLEncoder.encode(rule.getQuery() + " AND " + rule.getField() + ":\"" + entry.getKey() + "\"", "UTF-8");
                sb.append("Search: " + emailConfiguration.getWebInterfaceUri() + search_uri + "\n");

            }
        }
        return sb.toString();

    }

    public static Map<String, Object> parametersFromRule(Rule rule){
        String query = rule.getQuery();
        String streamId = rule.getStreamId();

        Map<String, Object> parameters = new HashMap<String, Object>();
        parameters.put("time", rule.getInterval());
        parameters.put("description", AggregatesUtil.getAlertConditionDescription(rule));

        if (rule.isMatchMoreOrEqual()){
            parameters.put("threshold_type", AggregatesAlertCondition.ThresholdType.MORE_OR_EQUAL.toString());
        } else {
            parameters.put("threshold_type", AggregatesAlertCondition.ThresholdType.LESS.toString());
        }

        parameters.put("threshold", rule.getNumberOfMatches());
        parameters.put("grace", 0);
        parameters.put("type", AggregatesUtil.ALERT_CONDITION_TYPE);
        parameters.put("field", rule.getField());
        parameters.put("number_of_matches", rule.getNumberOfMatches());
        parameters.put("match_more_or_equal", rule.isMatchMoreOrEqual());
        parameters.put("backlog", rule.getBacklog());
        parameters.put("repeat_notifications", rule.shouldRepeatNotifications());
        parameters.put("interval", rule.getInterval());
        parameters.put("query", query);
        parameters.put("rule_name", rule.getName());

        return parameters;
    }

    public static String alertConditionTitleFromRule(Rule rule){
        return "Aggregate rule [" + rule.getName() + "] triggered an alert.";
    }
} public class AuditEventTypes implements PluginAuditEventTypes {

    public static final String NAMESPACE = "aggregates:";

    public static final String AGGREGATES_RULE_CREATE = NAMESPACE + "aggregates_rule:create";
    public static final String AGGREGATES_RULE_UPDATE = NAMESPACE + "aggregates_rule:update";
    public static final String AGGREGATES_RULE_DELETE = NAMESPACE + "aggregates_rule:delete";
    public static final String AGGREGATES_REPORT_SCHEDULE_CREATE = NAMESPACE + "aggregates_report_schedule:create";
    public static final String AGGREGATES_REPORT_SCHEDULE_UPDATE = NAMESPACE + "aggregates_report_schedule:update";
    public static final String AGGREGATES_REPORT_SCHEDULE_DELETE = NAMESPACE + "aggregates_report_schedule:delete";

    private static final Set<String> EVENT_TYPES = ImmutableSet .<String>builder()
            .add(AGGREGATES_RULE_CREATE)
            .add(AGGREGATES_RULE_UPDATE)
            .add(AGGREGATES_RULE_DELETE)
            .add(AGGREGATES_REPORT_SCHEDULE_CREATE)
            .add(AGGREGATES_REPORT_SCHEDULE_UPDATE)
            .add(AGGREGATES_REPORT_SCHEDULE_DELETE)
            .build();

    @Override
    public Set<String> auditEventTypes() {
        return EVENT_TYPES;
    }
}
import org.graylog.plugins.aggregates.alerts.AggregatesAlertSender;
import org.graylog.plugins.aggregates.alarmcallbacks.AggregatesEmailAlarmCallback;
import org.graylog.plugins.aggregates.alerts.AggregatesAlertCondition;
import org.graylog.plugins.aggregates.alerts.FormattedEmailAlertSender;
import org.graylog.plugins.aggregates.history.HistoryItemService;
import org.graylog.plugins.aggregates.history.HistoryItemServiceImpl;
import org.graylog.plugins.aggregates.maintenance.AggregatesMaintenance;
import org.graylog.plugins.aggregates.permissions.RuleRestPermissions;
import org.graylog.plugins.aggregates.permissions.ReportScheduleRestPermissions;
import org.graylog.plugins.aggregates.report.AggregatesReport;
import org.graylog.plugins.aggregates.rule.RuleService;
import org.graylog.plugins.aggregates.rule.RuleServiceImpl;
import org.graylog.plugins.aggregates.rule.rest.RuleResource;
import org.graylog.plugins.aggregates.report.schedule.ReportScheduleService;
import org.graylog.plugins.aggregates.report.schedule.ReportScheduleServiceImpl;
import org.graylog.plugins.aggregates.report.schedule.rest.ReportScheduleResource;
import org.graylog.plugins.aggregates.util.AggregatesUtil;
import org.graylog.plugins.aggregates.util.AuditEventTypes;
import org.graylog2.alerts.AlertService;
import org.graylog2.alerts.AlertServiceImpl;
import org.graylog2.plugin.PluginConfigBean;
import org.graylog2.plugin.PluginModule;
import java.util.Collections;
import java.util.Set;
package org.graylog.plugins.aggregates;



/**
 * Extend the PluginModule abstract class here to add you plugin to the system.
 */
public class AggregatesModule extends PluginModule {
    /**
     * Returns all configuration beans required by this plugin.
     *
     * Implementing this method is optional. The default method returns an empty {@link Set}.
     */
    @Override
    public Set<? extends PluginConfigBean> getConfigBeans() {
        return Collections.emptySet();
    }

    @Override
    protected void configure() {
        bind(AlertService.class).to(AlertServiceImpl.class);
    	bind(RuleService.class).to(RuleServiceImpl.class);
    	bind(ReportScheduleService.class).to(ReportScheduleServiceImpl.class);