/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF 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.apache.camel.component.influxdb2;

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.WriteApi;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.client.write.Point;
import org.apache.camel.Exchange;
import org.apache.camel.InvalidPayloadException;
import org.apache.camel.component.influxdb2.data.Measurement;
import org.apache.camel.component.influxdb2.data.Measurements;
import org.apache.camel.component.influxdb2.data.Points;
import org.apache.camel.component.influxdb2.data.Record;
import org.apache.camel.component.influxdb2.data.Records;
import org.apache.camel.support.DefaultProducer;
import org.apache.camel.support.MessageHelper;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class InfluxDb2Producer extends DefaultProducer {
    private static final Logger LOG = LoggerFactory.getLogger(InfluxDb2Producer.class);
    InfluxDb2Endpoint endpoint;

    InfluxDBClient connection;

    WriteApi writeApi;

    public InfluxDb2Producer(InfluxDb2Endpoint endpoint) {
        super(endpoint);
        this.connection = endpoint.getInfluxDBClient();
        this.endpoint = endpoint;
        this.writeApi = connection.makeWriteApi();
    }

    /**
     * Processes the message exchange
     *
     * @param  exchange  the message exchange
     * @throws Exception if an internal processing error has occurred.
     */
    @Override
    public void process(Exchange exchange) throws Exception {
        String orgName = calculateOrgName(exchange);
        String bucketName = calculateBucketName(exchange);
        WritePrecision writePrecision = calculateWritePrecision(exchange);
        switch (endpoint.getOperation()) {
            case INSERT:
                doInsert(exchange, orgName, bucketName, writePrecision);
                break;
            case PING:
                doPing(exchange);
                break;
            default:
                throw new IllegalArgumentException("The operation " + endpoint.getOperation() + " is not supported");
        }
    }

    private void doInsert(Exchange exchange, String orgName, String bucketName, WritePrecision writePrecision)
            throws InvalidPayloadException {
        Object body = exchange.getIn().getBody();
        if (body instanceof Point) {
            insertPoint(exchange, orgName, bucketName);
        } else if (body instanceof Measurement) {
            insertMeasurement(exchange, orgName, bucketName, writePrecision);
        } else if (body instanceof Record) {
            insertRecord(exchange, orgName, bucketName, writePrecision);
        } else if (body instanceof Points) {
            insertPoints(exchange, orgName, bucketName);
        } else if (body instanceof Measurements) {
            insertMeasurements(exchange, orgName, bucketName, writePrecision);
        } else if (body instanceof Records) {
            insertRecords(exchange, orgName, bucketName, writePrecision);
        } else {
            // default insert as point
            insertPoint(exchange, orgName, bucketName);
        }
    }

    private void insertPoint(Exchange exchange, String orgName, String bucketName)
            throws InvalidPayloadException {
        Point point = exchange.getIn().getMandatoryBody(Point.class);
        try {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Writing point {}", point.toLineProtocol());
            }
            writeApi.writePoint(bucketName, orgName, point);
        } catch (Exception ex) {
            exchange.setException(new CamelInfluxDb2Exception(ex));
        }
    }

    private void insertMeasurement(Exchange exchange, String orgName, String bucketName, WritePrecision writePrecision)
            throws InvalidPayloadException {
        Measurement measurement = exchange.getIn().getMandatoryBody(Measurement.class);
        try {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Writing measurement {}", measurement);
            }
            writeApi.writeMeasurement(bucketName, orgName, writePrecision, measurement.getInfluxMeasurement());
        } catch (Exception ex) {
            exchange.setException(new CamelInfluxDb2Exception(ex));
        }
    }

    private void insertRecord(Exchange exchange, String orgName, String bucketName, WritePrecision writePrecision)
            throws InvalidPayloadException {
        Record recordObj = exchange.getIn().getMandatoryBody(Record.class);
        try {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Writing record {}", recordObj);
            }
            writeApi.writeRecord(bucketName, orgName, writePrecision, recordObj.getInfluxRecord());
        } catch (Exception ex) {
            exchange.setException(new CamelInfluxDb2Exception(ex));
        }
    }

    private void insertPoints(Exchange exchange, String orgName, String bucketName)
            throws InvalidPayloadException {
        @SuppressWarnings("unchecked")
        Points points = exchange.getIn().getMandatoryBody(Points.class);
        try {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Writing points {}", points);
            }
            writeApi.writePoints(bucketName, orgName, points.getInfluxPoints());
        } catch (Exception ex) {
            exchange.setException(new CamelInfluxDb2Exception(ex));
        }
    }

    @SuppressWarnings({ "unchecked", "rawtypes" })
    private void insertMeasurements(Exchange exchange, String orgName, String bucketName, WritePrecision writePrecision)
            throws InvalidPayloadException {
        Measurements measurements = exchange.getIn().getMandatoryBody(Measurements.class);
        try {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Writing measurements {}", measurements);
            }
            writeApi.writeMeasurements(bucketName, orgName, writePrecision, measurements.getInfluxMeasurements());
        } catch (Exception ex) {
            exchange.setException(new CamelInfluxDb2Exception(ex));
        }
    }

    private void insertRecords(Exchange exchange, String orgName, String bucketName, WritePrecision writePrecision)
            throws InvalidPayloadException {
        Records records = exchange.getIn().getMandatoryBody(Records.class);
        try {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Writing records {}", records);
            }
            writeApi.writeRecords(bucketName, orgName, writePrecision, records.getInfluxRecords());
        } catch (Exception ex) {
            exchange.setException(new CamelInfluxDb2Exception(ex));
        }
    }

    private void doPing(Exchange exchange) {
        Boolean result = connection.ping();
        MessageHelper.copyHeaders(exchange.getIn(), exchange.getOut(), true);
        exchange.getMessage().setBody(result);
    }

    private String calculateOrgName(Exchange exchange) {
        String orgName = exchange.getIn().getHeader(InfluxDb2Constants.ORG, String.class);
        if (ObjectHelper.isNotEmpty(orgName)) {
            return orgName;
        }
        return endpoint.getOrg();
    }

    private String calculateBucketName(Exchange exchange) {
        String bucketName = exchange.getIn().getHeader(InfluxDb2Constants.BUCKET, String.class);
        if (ObjectHelper.isNotEmpty(bucketName)) {
            return bucketName;
        }
        return endpoint.getBucket();
    }

    private WritePrecision calculateWritePrecision(Exchange exchange) {
        WritePrecision precision = exchange.getIn().getHeader(InfluxDb2Constants.WRITE_PRECISION, WritePrecision.class);
        if (ObjectHelper.isNotEmpty(precision)) {
            return precision;
        }
        return endpoint.getWritePrecision();
    }
}
