/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at
 * docs/licenses/cddl.txt
 * or http://www.opensource.org/licenses/cddl1.php.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at
 * docs/licenses/cddl.txt.  If applicable,
 * add the following below this CDDL HEADER, with the fields enclosed
 * by brackets "[]" replaced with your own identifying information:
 *      Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 *
 *
 *      Copyright 2018-2021 Ping Identity Corporation
 */
package com.unboundid.directory.sdk.sync.api;

import com.unboundid.directory.sdk.common.internal.ExampleUsageProvider;
import com.unboundid.directory.sdk.common.internal.Reconfigurable;
import com.unboundid.directory.sdk.common.internal.UnboundIDExtension;
import com.unboundid.directory.sdk.sync.config.ChangeDetectorConfig;
import com.unboundid.directory.sdk.sync.internal.SynchronizationServerExtension;
import com.unboundid.directory.sdk.sync.types.ChangeRecord;
import com.unboundid.directory.sdk.sync.types.EndpointException;
import com.unboundid.directory.sdk.sync.types.SetStartpointOptions;
import com.unboundid.directory.sdk.sync.types.SyncOperation;
import com.unboundid.directory.sdk.sync.types.SyncServerContext;
import com.unboundid.directory.sdk.sync.types.SyncSourceContext;
import com.unboundid.ldap.sdk.ResultCode;
import com.unboundid.util.Extensible;
import com.unboundid.util.args.ArgumentException;
import com.unboundid.util.args.ArgumentParser;

import java.io.Serializable;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

/**
 * This class defines an API that must be implemented by extensions that
 * detect changes for an LDAP based Sync Source.
 * A Change Detector can be used to
 * <ul>
 * <li>Process logs or other flat files for changes.</li>
 * <li>Process changes from a Queue (Kafka, RabbitMQ, etc)</li>
 * <li>Override the standard cn=changelog
 * based approach for detecting changes.</li>
 * </ul>
 * <br>
 * <h2>Configuring Change Detectors</h2>
 * In order to configure a Change Detector created using this API, use a
 * command like:
 * <pre>
 *    dsconfig create-change-detector \
 *          --detector-name "<i>{detector-name}</i>" \
 *          --type third-party \
 *          --set "extension-class:<i>{class-name}</i>" \
 *          --set "extension-argument:<i>{name=vale}</i>"
 * </pre>
 * where "<i>{plugin-name}</i>" is the name to use for the Change Detector
 * instance, "<i>{class-name}</i>" is the fully-qualified name of the Java
 * class that extends
 * {@code com.unboundid.directory.sdk.sync.api.ChangeDetector},
 * and "<i>{name=value}</i>" represents name-value pairs for any arguments to
 * provide to the Change Detector. If multiple arguments should be provided
 * to the Change Detector, then the
 * "<CODE>--set extension-argument:<I>{name=value}</I></CODE>"
 * option should be provided multiple times.
 */
@Extensible()
@SynchronizationServerExtension(appliesToLocalContent = false,
        appliesToSynchronizedContent = true)
public abstract class ChangeDetector
        implements UnboundIDExtension,
        Reconfigurable<ChangeDetectorConfig>,
        ExampleUsageProvider {

  /**
   * Creates a new instance of this LDAP Change Detector.  All Change
   * Detector implementations must include a default constructor, but any
   * initialization should generally be done in the
   * {@code initializeChangeDetector} method.
   */
  public ChangeDetector() {
    // No implementation is required.
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public abstract String getExtensionName();

  /**
   * {@inheritDoc}
   */
  @Override
  public abstract String[] getExtensionDescription();

  /**
   * {@inheritDoc}
   */
  @Override
  public Map<List<String>, String> getExamplesArgumentSets() {
    return Collections.emptyMap();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void defineConfigArguments(final ArgumentParser parser)
          throws ArgumentException {
    // No arguments will be allowed by default.
  }

  /**
   * This hook is called when a Sync Pipe first starts up  or when the
   * set-startpoint subcommand is called from the <i>realtime-sync</i> command
   * line tool. Any initialization of this change detector should be performed
   * here. This method should generally store the {@link SyncServerContext}
   * and {@link SyncSourceContext} in a class member so that it can be used
   * elsewhere in the implementation.
   * <p>
   * The default implementation is empty.
   *
   * @param serverContext     A handle to the server context for the server in
   *                          which this extension is running.
   * @param syncSourceContext An interface for interacting with the Sync
   *                          Source that owns this Change Detector or
   *                          {@code null} if the Change Detector is only
   *                          being initialized to validate its configuration.
   * @param parser            The argument parser which has been initialized
   *                          from the configuration for this sync source.
   */
  public void initializeChangeDetector(
          final SyncServerContext serverContext,
          final SyncSourceContext syncSourceContext,
          final ArgumentParser parser) {
    // No initialization will be performed by default.
  }

  /**
   * This hook is called when a Sync Pipe shuts down or when the set-startpoint
   * subcommand (from the <i>realtime-sync</i> command line tool) is finished.
   * Any clean up of this change detector should be performed here.
   * <p>
   * The default implementation is empty.
   */
  public void finalizeChangeDetector() {
    //No implementation required by default.
  }

  /**
   * {@inheritDoc}
   */
  public boolean isConfigurationAcceptable(
          final ChangeDetectorConfig config,
          final ArgumentParser parser,
          final List<String> unacceptableReasons) {
    // No implementation required by default.
    return true;
  }

  /**
   * {@inheritDoc}
   */
  public ResultCode applyConfiguration(final ChangeDetectorConfig config,
                                       final ArgumentParser parser,
                                       final List<String> adminActionsRequired,
                                       final List<String> messages) {
    // No implementation required by default.
    return ResultCode.SUCCESS;
  }

  /**
   * This method should effectively set the starting point for synchronization
   * to the place specified by the <code>options</code> parameter. This should
   * cause all changes previous to the specified start point to be disregarded
   * and only changes after that point to be returned by
   * {@link #getNextBatchOfChanges(int, AtomicLong)}.
   * <p>
   * There are several different startpoint types (see
   * {@link SetStartpointOptions}), and this implementation is not required to
   * support them all. If the specified startpoint type is unsupported, this
   * method should throw an {@link UnsupportedOperationException}.
   *
   * <b>IMPORTANT</b>: The <code>RESUME_AT_SERIALIZABLE</code> startpoint type
   * must be supported by your implementation, because this is used when a Sync
   * Pipe first starts up. The {@link Serializable} in this case is the same
   * type that is returned by {@link #getStartpoint()}; the Sync Server persists
   * it and passes it back in on a restart.
   * <p>
   * This method can be called from two different contexts:
   * <ul>
   * <li>When the 'set-startpoint' subcommand of the realtime-sync CLI is used
   * (the Sync Pipe is required to be stopped in this context)</li>
   * <li>Immediately after a Sync Pipe starts up and a connection is first
   * established to the source server (e.g. before the first call to
   * {@link #getNextBatchOfChanges(int, AtomicLong)})</li>
   * </ul>
   *
   * @param options an object which indicates where exactly to start
   *                synchronizing (e.g. the end of the changelog, specific
   *                change number, a certain time ago, etc)
   * @throws EndpointException if there is any error while setting the
   *                           start point
   */
  public abstract void setStartpoint(final SetStartpointOptions options)
          throws EndpointException;


  /**
   * Gets the current value of the startpoint for change detection. This is the
   * "bookmark" which indicates which changes have already been processed and
   * which have not. In most cases, a change number is used to detect changes
   * and is managed by the Data Sync Server, in which case this
   * implementation needs only to return the latest acknowledged
   * change number. In other cases, the return value may correspond to a
   * different value, such as the SYS_CHANGE_VERSION in Microsoft SQL Server.
   * In any case, this method should return the value that is updated by
   * {@link #acknowledgeCompletedOps(LinkedList)}.
   * <p>
   * This method is called periodically and the return value is saved in the
   * persistent state for the Sync Pipe that uses this extension as its Sync
   * Source.
   *
   * <b>IMPORTANT</b>: The internal value for the startpoint should only be
   * updated after a sync operation is acknowledged back to this extension (via
   * {@link #acknowledgeCompletedOps(LinkedList)}).
   * Otherwise it will be possible for changes to be missed when the
   * Data Sync Server is restarted or a connection error occurs.
   *
   * @return a value to store in the persistent state for the Sync Pipe. This is
   * usually a change number, but if a changelog table is not used to
   * detect changes, this value should represent some other token to
   * pass into {@link #setStartpoint(SetStartpointOptions)}
   * when the sync pipe starts up.
   */
  public abstract Serializable getStartpoint();

  /**
   * Return the next batch of change records from the source. Change records
   * are usually just hints that a change happened; they do not include
   * the full contents of the target entry. In an effort to never synchronize
   * stale data, the Data Sync Server will go back and fetch the full
   * target entry for each change record.
   * <p>
   * On the first invocation, this should return changes starting from the
   * startpoint that was set by
   * {@link #setStartpoint(SetStartpointOptions)}. This method is also
   * responsible for updating the internal state such that subsequent
   * invocations do not return duplicate changes.
   * <p>
   * The resulting list should be limited by <code>maxChanges</code>. The
   * <code>numStillPending</code> reference should be set to the estimated
   * number of changes that haven't yet been retrieved from the source endpoint
   * when this method returns, or zero if all the current changes have been
   * retrieved.
   *
   * <b>IMPORTANT</b>: While this method needs to keep track of which changes
   * have already been returned so that it does not return them again, it should
   * <b>NOT</b> modify the official startpoint. The internal value for the
   * startpoint should only be updated after a sync operation is acknowledged
   * back to this extension (via
   * {@link #acknowledgeCompletedOps(LinkedList)}).
   * Otherwise it will be possible for changes to be missed when the
   * Data Sync Server is restarted or a connection error occurs. The
   * startpoint should not change as a result of this method.
   * <p>
   * This method <b>does not need to be thread-safe</b>. It will be invoked
   * repeatedly by a single thread, based on the polling interval set in the
   * Sync Pipe configuration.
   *
   * @param maxChanges      the maximum number of changes to retrieve
   * @param numStillPending this should be set to the number of unretrieved
   *                        changes that are still pending after this batch has
   *                        been retrieved. This will be passed in as zero, and
   *                        may be left that way if the actual value cannot be
   *                        determined.
   * @return a list of {@link ChangeRecord} instances, each
   * corresponding to a single change at the source endpoint.
   * If there are no new changes to return, this method should return
   * an empty list.
   * @throws EndpointException if there is any error while retrieving the
   *                           next batch of changes
   */
  public abstract List<ChangeRecord> getNextBatchOfChanges(
          final int maxChanges,
          final AtomicLong numStillPending)
          throws EndpointException;

  /**
   * Provides a way for the Data Sync Server to acknowledge back to the
   * extension which sync operations it has processed. This method should update
   * the official startpoint which was set by
   * {@link #setStartpoint(SetStartpointOptions)} and is
   * returned by {@link #getStartpoint()}.
   *
   * <b>IMPORTANT</b>: The internal value for the startpoint should only be
   * updated after a sync operation is acknowledged back to this extension (via
   * this method). Otherwise it will be possible for changes to be missed when
   * the Data Sync Server is restarted or a connection error occurs.
   *
   * @param completedOps a list of {@link SyncOperation}s that have finished
   *                     processing. The records are listed in the order they
   *                     were first detected.
   * @throws EndpointException if there is an error acknowledging the changes
   *                           back to the source
   */
  public abstract void acknowledgeCompletedOps(
          final LinkedList<SyncOperation> completedOps)
          throws EndpointException;
}