/*******************************************************************************
 * Copyright 2018 572682
 * 
 * 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 us.dot.its.jpo.ode.model;

import java.util.UUID;

import com.fasterxml.jackson.databind.JsonNode;

public class SerialId {
   private static final char UUID_DELIMITER = '_';
   private static final char BUNDLE_RECORD_DELIMITER = '.';
   private static final char SERIAL_NUMBER_DELIMITER = '#';
   
   private String streamId;
   private int bundleSize = 1;
   private long bundleId = 0;
   private int recordId = 0;
   private long serialNumber = 0;

   
   public SerialId() {
      streamId = UUID.randomUUID().toString();
   }

   public SerialId(String streamId, int bundleSize, 
         long bundleId, int recordId) {
      this();
      if (streamId != null)
         this.streamId = streamId;
      else
         this.streamId = this.streamId + "_null";
      
      this.bundleSize = bundleSize;
      this.bundleId = bundleId + (recordId / this.bundleSize);
      this.recordId = recordId % bundleSize;
      this.serialNumber = calculateSerialNumber();
   }

   public SerialId (String serialId) throws Exception {
      
      String[] splitId  = serialId.split(
            "[" + UUID_DELIMITER + SERIAL_NUMBER_DELIMITER + BUNDLE_RECORD_DELIMITER +"]+");
      
      if (splitId.length != 5) {
         throw new Exception("Invalid serialId! Expected length 5 but was " + splitId.length);
      }
      
      this.streamId = splitId[0];
      this.bundleSize = Integer.parseInt(splitId[1]);
      this.bundleId = Integer.parseInt(splitId[2]);
      this.recordId = Integer.parseInt(splitId[3]);
      this.serialNumber = Integer.parseInt(splitId[4]);
   }
   
   public SerialId(String streamId, 
         int bundleSize, long bundleId, int recordId,
         long serialNumber) {
      
      this.streamId = streamId;
      this.bundleSize = bundleSize;
      this.bundleId = bundleId;
      this.recordId = recordId;
      this.serialNumber = serialNumber;
   }

   public SerialId(JsonNode jsonNode) {
      this(jsonNode.get("streamId").asText(), 
           jsonNode.get("bundleSize").asInt(), 
           jsonNode.get("bundleId").asLong(), 
           jsonNode.get("recordId").asInt(), 
           jsonNode.get("serialNumber").asLong());
   }

   private long calculateSerialNumber() {
      return (this.bundleId * this.bundleSize) + this.recordId;
   }

   public int nextRecordId() {
      return (recordId + 1) % bundleSize;
   }
   
   public long nextSerialNumber() {
      return serialNumber + 1;
   }
   
   public SerialId nextSerialId() {
      SerialId next = clone();
      next.increment();
      return next; 
   }
   
   synchronized public long increment() {
      bundleId += (recordId + 1) / bundleSize;
      recordId = nextRecordId();
      return ++serialNumber;
   }
   
   public SerialId clone() {
      SerialId clone = new SerialId(streamId, bundleSize, bundleId, recordId);
      clone.serialNumber = this.serialNumber;
      return clone;
   }
   

   public boolean isRightAfter (SerialId prev) {
      return (this.getSerialNumber() == prev.getSerialNumber() + 1);
   }
   
   public boolean isRightBefore (SerialId next) {
      return (this.getSerialNumber() + 1 == next.getSerialNumber());
   }
   
   public String getStreamId() {
      return streamId;
   }

   public SerialId setStreamId(String streamId) {
      this.streamId = streamId;
      return this;
   }

   public int getBundleSize() {
      return bundleSize;
   }

   public SerialId setBundleSize(int bundleSize) {
      this.bundleSize = bundleSize;
      return this;
   }

   public long getBundleId() {
      return bundleId;
   }

   public SerialId setBundleId(long bundleId) {
      this.bundleId = bundleId;
      return this;
   }

   public SerialId addBundleId(long num) {
       this.bundleId += num;
       return this;
    }

   public int getRecordId() {
      return recordId;
   }

   public SerialId setRecordId(int recordId) {
       this.recordId = recordId;
       return this;
    }

   public SerialId addRecordId(int num) {
       this.recordId += num;
       return this;
    }

   public long getSerialNumber() {
      return serialNumber;
   }

   @Override
   public String toString() {
      return streamId + UUID_DELIMITER + bundleSize + 
            BUNDLE_RECORD_DELIMITER + bundleId + 
            BUNDLE_RECORD_DELIMITER + recordId +
            SERIAL_NUMBER_DELIMITER + serialNumber;
   }

   @Override
   public int hashCode() {
      final int prime = 31;
      int result = 1;
      result = prime * result + (int) (bundleId ^ (bundleId >>> 32));
      result = prime * result + bundleSize;
      result = prime * result + recordId;
      result = prime * result + ((streamId == null) ? 0 : streamId.hashCode());
      return result;
   }

   @Override
   public boolean equals(Object obj) {
      if (this == obj)
         return true;
      if (obj == null)
         return false;
      if (getClass() != obj.getClass())
         return false;
      SerialId other = (SerialId) obj;
      if (bundleId != other.bundleId)
         return false;
      if (bundleSize != other.bundleSize)
         return false;
      if (recordId != other.recordId)
         return false;
      if (streamId == null) {
         if (other.streamId != null)
            return false;
      } else if (!streamId.equals(other.streamId))
         return false;
      return true;
   }

}
