/*
 * Copyright 2020 Google LLC
 *
 * 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
 *
 *     https://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 com.google.spez.core;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import com.google.spez.spanner.Row;
import java.util.Map;

@SuppressWarnings("PMD.BeanMembersShouldSerialize")
public class MetadataExtractor {

  private final SpezConfig config;
  private final ImmutableMap<String, String> base;

  public MetadataExtractor(SpezConfig config) {
    this.config = config;
    this.base = ImmutableMap.copyOf(config.getBaseMetadata());
  }

  /**
   * extract metadata for a given row.
   *
   * @param row metadata is retrieved from this
   * @return a map of metadata key value pairs
   */
  public Map<String, String> extract(Row row) {
    Map<String, String> metadata = Maps.newHashMap(base);

    // TODO(xjdr): fix this
    String uuid = "";
    String type = config.getSink().getUuidColumnType();
    if (type.equals("STRING(MAX)")) {
      uuid = row.getString(config.getSink().getUuidColumn());
    } else if (type.equals("INT64")) {
      uuid = String.valueOf(row.getLong(config.getSink().getUuidColumn()));
    }
    String commitTimestamp = row.getTimestamp(config.getSink().getTimestampColumn()).toString();

    metadata.put(config.SINK_UUID_KEY, uuid);
    metadata.put(config.SINK_TIMESTAMP_KEY, commitTimestamp);

    return metadata;
  }
}

@@ -1342,36 +1342,8 @@
 );%0A%0A
-    // TODO(xjdr): fix this%0A
     
@@ -1357,13 +1357,8 @@
 uuid
- = %22%22
 ;%0A  
@@ -1411,16 +1411,87 @@
 Type();%0A
+    // Spez currently only supports UUIDs of type STRING(MAX) or INT64%0A
     if (
@@ -1503,27 +1503,21 @@
 equals(%22
-STRING(MAX)
+INT64
 %22)) %7B%0A  
@@ -1531,19 +1531,32 @@
 d = 
+String.valueOf(
 row.get
-Stri
+Lo
 ng(c
@@ -1587,16 +1587,17 @@
 olumn())
+)
 ;%0A    %7D 
@@ -1605,73 +1605,34 @@
 lse 
-if (type.equals(%22INT64%22)) %7B%0A      uuid = String.valueOf(
+%7B%0A      uuid = 
 row.get
-Lo
+Stri
 ng(c
@@ -1659,25 +1659,24 @@
 uidColumn())
-)
 ;%0A    %7D%0A    

move to only support INT64 and STRING(MAX) uuid types