/*
 * The contents of this file are subject to the terms of the Common Development and
 * Distribution License (the License). You may not use this file except in compliance with the
 * License.
 *
 * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
 * specific language governing permission and limitations under the License.
 *
 * When distributing Covered Software, include this CDDL Header Notice in each file and include
 * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
 * Header, with the fields enclosed by brackets [] replaced by your own identifying
 * information: "Portions copyright [year] [name of copyright owner]".
 *
 * Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.plugin;

import static java.nio.charset.StandardCharsets.*;
import static org.assertj.core.api.Assertions.*;
import static org.forgerock.opendj.ldap.ModificationType.*;
import static org.opends.messages.ReplicationMessages.*;
import static org.opends.server.TestCaseUtils.*;
import static org.opends.server.core.DirectoryServer.*;
import static org.opends.server.protocols.internal.InternalClientConnection.*;
import static org.testng.Assert.*;

import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.atomic.AtomicReference;

import org.forgerock.opendj.config.server.ConfigChangeResult;
import org.forgerock.opendj.ldap.DN;
import org.forgerock.opendj.ldap.ResultCode;
import org.forgerock.opendj.server.config.meta.ReplicationDomainCfgDefn.AssuredType;
import org.opends.server.TestCaseUtils;
import org.opends.server.core.ModifyOperation;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.AssuredMode;
import org.opends.server.replication.protocol.DoneMsg;
import org.opends.server.replication.protocol.EntryMsg;
import org.opends.server.replication.protocol.InitializeTargetMsg;
import org.opends.server.replication.server.ReplServerFakeConfiguration;
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.service.ReplicationBroker;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

/**
 * Tests a configuration change while this replica is the target of a total update.
 * <p>
 * The import of a total update streams over the session of the domain, on its listener
 * thread, and a configuration change which restarts that session for what it carries stops
 * the broker the import is reading (issue #1040). Through the server configuration the
 * change holds the lock of the configuration while it runs, and the restart waits for the
 * listener thread to end - which needs that lock to enable the backend back once the
 * stream ends: the change never returns. Reached below the configuration listeners, as the
 * change of an entry which was accepted before the import started reaches it, the restart
 * ends the import on the entries which had arrived.
 * <p>
 * The exporter is a broker of this test, so that the test says when the entries arrive:
 * the change is made while the import is waiting for them.
 */
@SuppressWarnings("javadoc")
public class ConfigChangeDuringImportTest extends ReplicationTestCase
{
  /**
   * The memory backend of {@code o=test} loses its data when it is disabled and enabled
   * back, which is what an import does to the backend it replaces: a total update needs a
   * backend which keeps what was imported into it.
   */
  private static final String EXAMPLE_DN = "dc=example,dc=com";
  private static final int RS_ID = 612;
  private static final int DS_ID = 1;
  private static final int EXPORTER_ID = 2;
  private static final int INIT_WINDOW = 100;
  private static final String DOMAIN_CONFIG_NAME = "config change during import test";
  private static final String IMPORTED_ENTRY_DN = "cn=imported,ou=People," + EXAMPLE_DN;
  /** How long a configuration change is given to return before it is read as hung. */
  private static final long CHANGE_TIMEOUT_IN_MS = 30_000;

  private DN baseDN;
  private int rsPort;
  private ReplicationServer replicationServer;
  private LDAPReplicationDomain domain;
  /** The entry the domain is configured in, when it is configured through the server. */
  private DN domainConfigDN;
  private ReplicationBroker exporter;

  @BeforeMethod
  public void setUpLocal() throws Exception
  {
    baseDN = DN.valueOf(EXAMPLE_DN);
    TestCaseUtils.clearBackend("userRoot", EXAMPLE_DN);

    rsPort = TestCaseUtils.findFreePort();
    replicationServer = new ReplicationServer(new ReplServerFakeConfiguration(
        rsPort, "configChangeDuringImportTestDb", 0, RS_ID, 0, 100, new TreeSet<String>()));
  }

  @AfterMethod(timeOut = 120_000)
  public void tearDown() throws Exception
  {
    try
    {
      stop(exporter);
      if (domainConfigDN != null)
      {
        // Deletes the "cn=external changelog" entry below it, and the domain, as well.
        deleteEntry(domainConfigDN);
        configEntriesToCleanup.remove(domainConfigDN);
        synchroServerEntry = null;
      }
      else if (domain != null)
      {
        MultimasterReplication.deleteDomain(baseDN);
      }
    }
    finally
    {
      exporter = null;
      domainConfigDN = null;
      domain = null;
      remove(replicationServer);
    }
  }

  /**
   * A change of the external changelog entry while the import streams must be refused, as
   * a change of the domain entry is.
   * <p>
   * The change is made the way {@code dsconfig} makes it, through the server configuration,
   * which holds the lock of the configuration for the length of it. Without the refusal the
   * attributes are applied and the session restarted for them: the restart stops the broker
   * the import is reading and waits for the listener thread, which ends the import on what
   * had arrived and then waits for the lock of the configuration to enable the backend back
   * - the change never returns, the backend stays deregistered, and every configuration
   * change of the server after it waits on the same lock.
   */
  @Test(timeOut = 180_000)
  public void aChangeOfTheExternalChangelogEntryIsRefusedWhileATotalUpdateRuns() throws Exception
  {
    configureDomainThroughTheServer();
    final String[] exported = exportedEntries();
    startImportInto(exported.length);

    final ModifyOperation change = changeConfigurationEntry(
        DN.valueOf("cn=external changelog," + domainConfigDN), "ds-cfg-ecl-include", "cn");

    assertEquals(change.getResultCode(), ResultCode.UNWILLING_TO_PERFORM,
        "a change of the external changelog entry was accepted while a total update ran: "
            + change.getErrorMessage());
    assertThat(change.getErrorMessage().toString())
        .as("the refusal does not say a total update is the reason")
        .contains(NOTE_ERR_CANNOT_CHANGE_CONFIG_DURING_TOTAL_UPDATE.get().toString());

    finishImport(exported);
    assertImported(exported);
    assertFalse(domain.getEclIncludes().contains("cn"),
        "the refused change of the attributes published to the external changelog was applied");
  }

  /**
   * A change of the domain entry while the import streams is refused, as it was before
   * this fix: the twin of the case above, pinned so that the two entries keep answering the
   * same thing.
   */
  @Test(timeOut = 180_000)
  public void aChangeOfTheDomainEntryIsRefusedWhileATotalUpdateRuns() throws Exception
  {
    configureDomainThroughTheServer();
    final String[] exported = exportedEntries();
    startImportInto(exported.length);

    final ModifyOperation change =
        changeConfigurationEntry(domainConfigDN, "ds-cfg-assured-type", "safe-read");

    assertEquals(change.getResultCode(), ResultCode.UNWILLING_TO_PERFORM,
        "a change of the domain entry was accepted while a total update ran: "
            + change.getErrorMessage());
    assertThat(change.getErrorMessage().toString())
        .as("the refusal does not say a total update is the reason")
        .contains(NOTE_ERR_CANNOT_CHANGE_CONFIG_DURING_TOTAL_UPDATE.get().toString());

    finishImport(exported);
    assertImported(exported);
    assertEquals(domain.getAssuredMode(), AssuredMode.SAFE_DATA_MODE,
        "the refused change of the assured configuration was applied");
  }

  /**
   * A change of the domain configuration which reaches the domain while the import streams
   * - one accepted before the import started - must leave the session to the import.
   * <p>
   * The assured configuration is negotiated as the session comes up, so the change asks for
   * a restart. The restart is refused and reported, the way it is for a domain disabled for
   * a total update: the configuration is stored, the import ends on the session it started
   * on, and the session the import brings up next negotiates what was stored.
   */
  @Test(timeOut = 180_000)
  public void aChangeOfTheDomainConfigurationLeavesTheSessionToTheImport() throws Exception
  {
    startDomain(domainCfg(AssuredType.NOT_ASSURED));
    final String[] exported = exportedEntries();
    startImportInto(exported.length);
    final Thread listener = listenerThread();
    assertNotNull(listener, "the import is running on no listener thread");

    final ConfigChangeResult ccr = domain.applyConfigurationChange(domainCfg(AssuredType.SAFE_READ));

    assertEquals(ccr.getResultCode(), ResultCode.SUCCESS, ccr.getMessages().toString());
    assertTrue(ccr.adminActionRequired(),
        "the change was reported as live although the session was not restarted for it");
    assertThat(ccr.getMessages().toString())
        .contains(NOTE_REPLICATION_DOMAIN_SESSION_NOT_RESTARTED.get(baseDN).toString());
    assertSame(listenerThread(), listener,
        "the session the import streams over was restarted for the change");

    finishImport(exported);
    assertImported(exported);
    assertEquals(domain.getAssuredMode(), AssuredMode.SAFE_READ_MODE,
        "the assured configuration was dropped although the change reported success");
  }

  /**
   * A change of the attributes published to the external changelog which reaches the domain
   * while the import streams must leave the session to the import: the attributes are
   * stored, and the session the import brings up next publishes them.
   */
  @Test(timeOut = 180_000)
  public void aChangeOfTheExternalChangelogAttributesLeavesTheSessionToTheImport() throws Exception
  {
    startDomain(domainCfg(AssuredType.NOT_ASSURED));
    final String[] exported = exportedEntries();
    startImportInto(exported.length);
    final Thread listener = listenerThread();
    assertNotNull(listener, "the import is running on no listener thread");

    final SortedSet<String> eclIncludes = new TreeSet<>();
    eclIncludes.add("cn");
    domain.changeConfig(eclIncludes, new TreeSet<String>());

    assertSame(listenerThread(), listener,
        "the session the import streams over was restarted for the change");

    finishImport(exported);
    assertImported(exported);
    assertTrue(domain.getEclIncludes().contains("cn"),
        "the attributes published to the external changelog were dropped");
  }

  /**
   * Configures the domain the way the server does, through its configuration entry: the
   * change listeners of that entry and of the "cn=external changelog" entry below it are
   * registered, and a change of either goes through the lock of the configuration.
   */
  private void configureDomainThroughTheServer() throws Exception
  {
    addSynchroServerEntry(
        "dn: cn=" + DOMAIN_CONFIG_NAME + ",cn=domains," + SYNCHRO_PLUGIN_DN + "\n"
        + "objectClass: top\n"
        + "objectClass: ds-cfg-replication-domain\n"
        + "cn: " + DOMAIN_CONFIG_NAME + "\n"
        + "ds-cfg-base-dn: " + EXAMPLE_DN + "\n"
        + "ds-cfg-replication-server: localhost:" + rsPort + "\n"
        + "ds-cfg-server-id: " + DS_ID + "\n"
        + "ds-cfg-assured-type: safe-data\n"
        + "ds-cfg-assured-sd-level: 1\n");
    domainConfigDN = synchroServerEntry.getName();
    assertTrue(getServerContext().getConfigurationHandler()
            .hasEntry(DN.valueOf("cn=external changelog," + domainConfigDN)),
        "the domain was configured without the external changelog entry this test changes");
    domain = MultimasterReplication.findDomain(baseDN, null);
    assertNotNull(domain, "the configuration entry created no domain");
    assertTrue(domain.isConnected(), "the domain did not connect to the replication server");
    exporter = openReplicationSession(baseDN, EXPORTER_ID, 100, rsPort, 10000);
  }

  private void startDomain(DomainFakeCfg cfg) throws Exception
  {
    domain = MultimasterReplication.createNewDomain(cfg);
    domain.start();
    assertTrue(domain.isConnected(), "the domain did not connect to the replication server");
    exporter = openReplicationSession(baseDN, EXPORTER_ID, 100, rsPort, 10000);
  }

  /**
   * A configuration which differs from the one the domain was started on by its assured
   * type alone: the broker properties are the same, so the change asks for no restart of
   * its own and what is left is the one the assured configuration asks for.
   */
  private DomainFakeCfg domainCfg(AssuredType assuredType)
  {
    final SortedSet<String> replServers = new TreeSet<>();
    replServers.add("localhost:" + rsPort);
    return new DomainFakeCfg(baseDN, DS_ID, replServers, assuredType, 1, -1, 1000, null);
  }

  /**
   * Changes a configuration entry the way {@code dsconfig} does, through the server
   * configuration, and fails when the change does not return: the restart it asks for
   * waits for the listener thread to end, which waits for the lock of the configuration the
   * change holds. The thread of the change is interrupted then, which gives up that wait
   * and lets the domain be taken down.
   */
  private ModifyOperation changeConfigurationEntry(DN entryDN, String attribute, String value)
      throws Exception
  {
    final AtomicReference<ModifyOperation> result = new AtomicReference<>();
    final Thread change = new Thread(new Runnable()
    {
      @Override
      public void run()
      {
        result.set(getRootConnection().processModify(
            modifyRequest(entryDN, REPLACE, attribute, value)));
      }
    }, "configuration change during the import");
    change.start();
    change.join(CHANGE_TIMEOUT_IN_MS);
    if (change.isAlive())
    {
      final String stacks = "\n" + stackOf(change) + "\n" + stackOf(listenerThread());
      change.interrupt();
      change.join(CHANGE_TIMEOUT_IN_MS);
      org.testng.Assert.fail("the change of " + entryDN + " did not return while the import ran:" + stacks);
    }
    return result.get();
  }

  private static String stackOf(Thread thread)
  {
    if (thread == null)
    {
      return "<no thread>";
    }
    final StringBuilder sb = new StringBuilder(thread.getName()).append(" [").append(thread.getState()).append("]\n");
    for (StackTraceElement frame : thread.getStackTrace())
    {
      sb.append("    at ").append(frame).append('\n');
    }
    return sb.toString();
  }

  /**
   * The listener thread of the domain is the one which says which session is running: the
   * import runs on it, and a restart of the session replaces it.
   */
  private Thread listenerThread()
  {
    final String name = "Replica DS(" + DS_ID + ") listener for domain \"" + baseDN + "\"";
    for (Thread thread : Thread.getAllStackTraces().keySet())
    {
      if (thread.getName().contains(name) && thread.isAlive())
      {
        return thread;
      }
    }
    return null;
  }

  /**
   * Has the exporter start a total update into this replica, and returns once the backend
   * of the domain is deregistered for it: from then on the import is reading the session.
   */
  private void startImportInto(int entryCount) throws Exception
  {
    exporter.publish(new InitializeTargetMsg(
        baseDN, EXPORTER_ID, DS_ID, EXPORTER_ID, entryCount, INIT_WINDOW));
    final long deadline = System.currentTimeMillis() + 30_000;
    while (getServerContext().getBackendConfigManager().findLocalBackendForEntry(baseDN) != null)
    {
      assertTrue(System.currentTimeMillis() < deadline,
          "the import did not deregister the backend of the domain");
      Thread.sleep(20);
    }
  }

  /** Has the exporter send the entries of the total update, and waits for the import to end. */
  private void finishImport(String... ldifEntries) throws Exception
  {
    int msgId = 0;
    for (String ldif : ldifEntries)
    {
      exporter.publish(new EntryMsg(EXPORTER_ID, DS_ID, ldif.getBytes(UTF_8), ++msgId));
    }
    exporter.publish(new DoneMsg(EXPORTER_ID, DS_ID));
    final long deadline = System.currentTimeMillis() + 60_000;
    while (domain.ieRunning())
    {
      assertTrue(System.currentTimeMillis() < deadline, "the import did not end");
      Thread.sleep(50);
    }
  }

  private static void assertImported(String... ldifEntries) throws Exception
  {
    for (String ldif : ldifEntries)
    {
      final DN dn = dnOf(ldif);
      assertTrue(entryExists(dn), "the import ended before " + dn
          + " arrived: the session it streams over was stopped from under it");
    }
  }

  /** The data of the exporter: the base entry and two entries below it. */
  private static String[] exportedEntries()
  {
    return new String[] {
      "dn: " + EXAMPLE_DN + "\n"
          + "objectClass: top\n"
          + "objectClass: domain\n"
          + "dc: example\n"
          + "entryUUID: 31111111-1111-1111-1111-111111111111\n"
          + "\n",
      "dn: ou=People," + EXAMPLE_DN + "\n"
          + "objectClass: top\n"
          + "objectClass: organizationalUnit\n"
          + "ou: People\n"
          + "entryUUID: 31111111-1111-1111-1111-111111111112\n"
          + "\n",
      "dn: " + IMPORTED_ENTRY_DN + "\n"
          + "objectClass: top\n"
          + "objectClass: person\n"
          + "cn: imported\n"
          + "sn: imported\n"
          + "entryUUID: 31111111-1111-1111-1111-111111111113\n"
          + "\n",
    };
  }

  private static DN dnOf(String ldif)
  {
    return DN.valueOf(ldif.substring("dn: ".length(), ldif.indexOf('\n')));
  }
}