/*
 * 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.beam.sdk.extensions.ordered;

import com.google.auto.value.AutoValue;
import java.util.Objects;
import javax.annotation.Nullable;
import org.apache.beam.sdk.schemas.AutoValueSchema;
import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
import org.joda.time.Instant;

/** Indicates the status of ordered processing for a particular key. */
@AutoValue
@DefaultSchema(AutoValueSchema.class)
public abstract class OrderedProcessingStatus {

  public static OrderedProcessingStatus create(
      @Nullable Long lastProcessedSequence,
      long numberOfBufferedEvents,
      @Nullable Long earliestBufferedSequence,
      @Nullable Long latestBufferedSequence,
      long numberOfReceivedEvents,
      long resultCount,
      long duplicateCount,
      boolean lastEventReceived,
      @Nullable ContiguousSequenceRange lastContiguousRange) {
    return new AutoValue_OrderedProcessingStatus.Builder()
        .setLastProcessedSequence(lastProcessedSequence)
        .setNumberOfBufferedEvents(numberOfBufferedEvents)
        .setEarliestBufferedSequence(earliestBufferedSequence)
        .setLatestBufferedSequence(latestBufferedSequence)
        .setNumberOfReceivedEvents(numberOfReceivedEvents)
        .setLastEventReceived(lastEventReceived)
        .setDuplicateCount(duplicateCount)
        .setResultCount(resultCount)
        .setStatusDate(Instant.now())
        .setLastContiguousSequenceRange(lastContiguousRange)
        .build();
  }

  /**
   * @return Last sequence processed. If null is returned - no elements for the given key and window
   *     have been processed yet.
   */
  public abstract @Nullable Long getLastProcessedSequence();

  /** @return Number of events received out of sequence and buffered. */
  public abstract long getNumberOfBufferedEvents();

  /** @return Earliest buffered sequence. If null is returned - there are no buffered events. */
  @Nullable
  public abstract Long getEarliestBufferedSequence();

  /** @return Latest buffered sequence. If null is returned - there are no buffered events. */
  @Nullable
  public abstract Long getLatestBufferedSequence();

  /** @return Total number of events received for the given key and window. */
  public abstract long getNumberOfReceivedEvents();

  /**
   * @return Number of duplicate events which were output in {@link
   *     OrderedEventProcessorResult#unprocessedEvents()} PCollection
   */
  public abstract long getDuplicateCount();

  /** @return Number of output results produced. */
  public abstract long getResultCount();

  /**
   * @return Indicator that the last event for the given key and window has been received. It
   *     doesn't necessarily mean that all the events for the given key and window have been
   *     processed. Use {@link OrderedProcessingStatus#getNumberOfBufferedEvents()} == 0 and this
   *     indicator as the sign that the processing is complete.
   */
  public abstract boolean isLastEventReceived();

  /**
   * @return Timestamp of when the status was produced. It is not related to the event's timestamp.
   */
  public abstract Instant getStatusDate();

  @Nullable
  public abstract ContiguousSequenceRange getLastContiguousSequenceRange();

  @Override
  public final boolean equals(@Nullable Object obj) {
    if (obj == null) {
      return false;
    }
    if (!OrderedProcessingStatus.class.isAssignableFrom(obj.getClass())) {
      return false;
    }
    OrderedProcessingStatus that = (OrderedProcessingStatus) obj;
    boolean result =
        Objects.equals(this.getEarliestBufferedSequence(), that.getEarliestBufferedSequence())
            && Objects.equals(this.getLastProcessedSequence(), that.getLastProcessedSequence())
            && Objects.equals(this.getLatestBufferedSequence(), that.getLatestBufferedSequence())
            && this.getNumberOfBufferedEvents() == that.getNumberOfBufferedEvents()
            && this.getDuplicateCount() == that.getDuplicateCount()
            && this.getResultCount() == that.getResultCount()
            && this.getNumberOfReceivedEvents() == that.getNumberOfReceivedEvents();
    return result;
  }

  @Override
  public final int hashCode() {
    return Objects.hash(
        this.getEarliestBufferedSequence(),
        this.getLastProcessedSequence(),
        this.getLatestBufferedSequence(),
        this.getNumberOfBufferedEvents(),
        this.getNumberOfReceivedEvents(),
        this.getDuplicateCount(),
        this.getResultCount());
  }

  @AutoValue.Builder
  public abstract static class Builder {

    public abstract Builder setLastProcessedSequence(@Nullable Long value);

    public abstract Builder setNumberOfBufferedEvents(long value);

    public abstract Builder setEarliestBufferedSequence(@Nullable Long value);

    public abstract Builder setLatestBufferedSequence(@Nullable Long value);

    public abstract Builder setNumberOfReceivedEvents(long value);

    public abstract Builder setDuplicateCount(long value);

    public abstract Builder setResultCount(long value);

    public abstract Builder setLastEventReceived(boolean value);

    public abstract Builder setStatusDate(Instant value);

    public abstract Builder setLastContiguousSequenceRange(@Nullable ContiguousSequenceRange value);

    public abstract OrderedProcessingStatus build();
  }
}
