/*
 * Copyright (c) 2017. Hans-Peter Grahsl (grahslhp@gmail.com)
 *
 * Licensed 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 at.grahsl.kafka.connect.mongodb.processor.field.projection;

import at.grahsl.kafka.connect.mongodb.MongoDbSinkConnectorConfig;
import com.mongodb.DBCollection;
import org.bson.BsonDocument;
import org.bson.BsonValue;

import java.util.Arrays;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

public abstract class WhitelistProjector extends FieldProjector {

    public WhitelistProjector(MongoDbSinkConnectorConfig config) {
        this(config, config.getValueProjectionList());
    }

    public WhitelistProjector(MongoDbSinkConnectorConfig config,
                              Set<String> fields) {
        super(config);
        this.fields = fields;
    }

    @Override
    protected void doProjection(String field, BsonDocument doc) {

        //special case short circuit check for '**' pattern
        //this is essentially the same as not using
        //whitelisting at all but instead take the full record
        if(fields.contains(FieldProjector.DOUBLE_WILDCARD)) {
            return;
        }

        Iterator<Map.Entry<String, BsonValue>> iter = doc.entrySet().iterator();
        while(iter.hasNext()) {
            Map.Entry<String, BsonValue> entry = iter.next();

            String key = field.isEmpty() ? entry.getKey()
                    : field + FieldProjector.SUB_FIELD_DOT_SEPARATOR + entry.getKey();
            BsonValue value = entry.getValue();

            if(!fields.contains(key)
                    //NOTE: always keep the _id field
                    && !key.equals(DBCollection.ID_FIELD_NAME)) {

                if(!checkForWildcardMatch(key))
                    iter.remove();

            }

            if(value.isDocument()) {
                //short circuit check to avoid recursion
                //if 'key.**' pattern exists
                String matchDoubleWildCard = key
                        + FieldProjector.SUB_FIELD_DOT_SEPARATOR
                        + FieldProjector.DOUBLE_WILDCARD;
                if(!fields.contains(matchDoubleWildCard)) {
                    doProjection(key, (BsonDocument)value);
                }
            }

        }
    }

    private boolean checkForWildcardMatch(String key) {

        String[] keyParts = key.split("\\"+FieldProjector.SUB_FIELD_DOT_SEPARATOR);
        String[] pattern = new String[keyParts.length];
        Arrays.fill(pattern,FieldProjector.SINGLE_WILDCARD);

        for(int c=(int)Math.pow(2, keyParts.length)-1;c >= 0;c--) {

            int mask = 0x1;
            for(int d = keyParts.length-1;d >= 0;d--) {
                if((c & mask) != 0x0) {
                    pattern[d] = keyParts[d];
                }
                mask <<= 1;
            }

            if(fields.contains(String.join(FieldProjector.SUB_FIELD_DOT_SEPARATOR,pattern)))
                return true;

            Arrays.fill(pattern,FieldProjector.SINGLE_WILDCARD);
        }

        return false;
    }
}

@@ -790,16 +790,43 @@
 ection;%0A
+import org.bson.BsonArray;%0A
 import o
@@ -2313,32 +2313,68 @@
 %0A            %7D%0A%0A
+            if(value != null) %7B%0A    
             if(v
@@ -2390,24 +2390,28 @@
 cument()) %7B%0A
+    
             
@@ -2471,16 +2471,20 @@
         
+ 
+   
 //if 'ke
@@ -2504,16 +2504,20 @@
  exists%0A
+    
         
@@ -2573,32 +2573,36 @@
                 
+    
 + FieldProjector
@@ -2642,32 +2642,36 @@
                 
+  
+  
 + FieldProjector
@@ -2688,16 +2688,20 @@
 LDCARD;%0A
+    
         
@@ -2772,16 +2772,20 @@
         
+ 
+   
 doProjec
@@ -2816,16 +2816,380 @@
 value);%0A
+                    %7D%0A                %7D%0A                if(value.isArray()) %7B%0A                    BsonArray values = (BsonArray)value;%0A                    for(BsonValue v : values.getValues()) %7B%0A                        if(v != null && v.isDocument()) %7B%0A                            doProjection(key,(BsonDocument)v);%0A                        %7D%0A                    %7D%0A
         

allow for whitelist projection for fields of documents within arrays