// ============================================================================
//
// Copyright (C) 2006-2017 Talend Inc. - www.talend.com
//
// This source code is available under agreement available at
// %InstallDIR%\features\org.talend.rcp.branding.%PRODUCTNAME%\%PRODUCTNAME%license.txt
//
// You should have received a copy of the agreement
// along with this program; if not, write to Talend SA
// 9 rue Pages 92150 Suresnes, France
//
// ============================================================================
package org.talend.components.salesforce.runtime;

import java.util.ArrayList;
import java.util.List;

import org.apache.avro.Schema;
import org.apache.avro.Schema.Field;
import org.apache.avro.generic.IndexedRecord;
import org.talend.components.api.component.runtime.Result;
import org.talend.components.api.component.runtime.WriteOperation;
import org.talend.components.api.container.RuntimeContainer;
import org.talend.components.common.BulkFileProperties;
import org.talend.components.common.runtime.BulkFileWriter;
import org.talend.components.salesforce.SalesforceOutputProperties.OutputAction;
import org.talend.components.salesforce.tsalesforceoutputbulk.TSalesforceOutputBulkProperties;
import org.talend.daikon.avro.SchemaConstants;

/**
 * Prepare Data Files for bulk execution
 */
final class SalesforceBulkFileWriter extends BulkFileWriter {

    public SalesforceBulkFileWriter(WriteOperation<Result> writeOperation, BulkFileProperties bulkProperties,
            RuntimeContainer adaptor) {
        super(writeOperation, bulkProperties, adaptor);
    }

    @Override
    public String[] getHeaders(Schema schema) {
        TSalesforceOutputBulkProperties salesforceBulkProperties = getBulkProperties();

        List<String> headers = new ArrayList<String>();
        StringBuilder sbuilder = new StringBuilder();
        for (Schema.Field f : schema.getFields()) {
            String header = f.name();
            if (checkDeleteOption(f)) {
                continue;
            }
            String ref_module_name = f.getProp(SalesforceSchemaConstants.REF_MODULE_NAME);
            String ref_field_name = f.getProp(SalesforceSchemaConstants.REF_FIELD_NAME);
            if (ref_module_name != null) {
                header = sbuilder.append(ref_module_name).append(":").append(ref_field_name).append(".").append(f.name())
                        .toString();
                sbuilder.setLength(0);
            } else {
                Object value = salesforceBulkProperties.upsertRelationTable.columnName.getValue();
                if (value != null && value instanceof List) {
                    int index = getIndex((List<String>) value, header);
                    if (index > -1) {
                        List<Boolean> polymorphics = salesforceBulkProperties.upsertRelationTable.polymorphic.getValue();
                        List<String> lookupFieldModuleNames = salesforceBulkProperties.upsertRelationTable.lookupFieldModuleName
                                .getValue();
                        List<String> lookupRelationshipFieldNames = salesforceBulkProperties.upsertRelationTable.lookupRelationshipFieldName
                                .getValue();
                        List<String> externalIdFromLookupFields = salesforceBulkProperties.upsertRelationTable.lookupFieldExternalIdName
                                .getValue();

                        Object polymorphic = polymorphics.get(index);
                        boolean poly = false;
                        if(polymorphic!=null && polymorphic instanceof Boolean) {
                            poly = (Boolean)polymorphic;
                        }

                        if (poly) {
                            sbuilder.append(lookupFieldModuleNames.get(index)).append(":");
                        }
                        sbuilder.append(lookupRelationshipFieldNames.get(index)).append(".")
                                .append(externalIdFromLookupFields.get(index));
                        header = sbuilder.toString();
                        sbuilder.setLength(0);
                    }
                }
            }

            headers.add(header);
        }
        return headers.toArray(new String[headers.size()]);
    }

    private int getIndex(List<String> columnNames, String columnName) {
        if (columnNames == null) {
            return -1;
        }
        return columnNames.indexOf(columnName);
    }

    private boolean checkDeleteOption(Field f) {
        return OutputAction.DELETE.equals(getBulkProperties().outputAction.getValue())
                && !Boolean.valueOf(f.getProp(SchemaConstants.TALEND_COLUMN_IS_KEY));
    }

    @Override
    public List<String> getValues(Object datum) {
        IndexedRecord input = getFactory(datum).convertToAvro((IndexedRecord) datum);
        List<String> values = new ArrayList<String>();
        for (Field f : input.getSchema().getFields()) {
            if (checkDeleteOption(f)) {
                continue;
            }
            if (input.get(f.pos()) == null) {
                if (getBulkProperties().ignoreNull.getValue()) {
                    values.add("");
                } else {
                    values.add("#N/A");
                }
            } else {
                values.add(String.valueOf(input.get(f.pos())));
            }
        }
        return values;
    }

    protected TSalesforceOutputBulkProperties getBulkProperties() {
        return (TSalesforceOutputBulkProperties) bulkProperties;
    }
}

@@ -1962,33 +1962,38 @@
 eckDeleteOption(
-f
+header
 )) %7B%0A           
@@ -4507,23 +4507,33 @@
 eOption(
-Field f
+String columnName
 ) %7B%0A    
@@ -4639,71 +4639,40 @@
 && !
-Boolean.valueOf(f.getProp(SchemaConstants.TALEND_COLUMN_IS_KEY)
+%22Id%22.equalsIgnoreCase(columnName
 );%0A 
@@ -4973,16 +4973,23 @@
 Option(f
+.name()
 )) %7B%0A   

fix issue with dynamic field. (#916)