/*
 * 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.protocol;

import static org.assertj.core.api.Assertions.assertThat;

import java.io.Closeable;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.opends.server.TestCaseUtils;
import org.opends.server.replication.ReplicationTestCase;
import org.opends.server.replication.common.CSN;
import org.opends.server.replication.common.CSNGenerator;
import org.opends.server.util.StaticUtils;
import org.testng.annotations.Test;

/**
 * A message published to a running session is handed to the session thread, which writes it to
 * the socket later. A caller which needs to know when that happened - the replication server
 * forwarding a ReplicaOfflineMsg, whose shutdown must not close the session before the message
 * is on the wire - attaches a callback to the message, and the session runs it once, from the
 * thread which wrote it, only after the write returned.
 * <p>
 * The peer of each test reads nothing until the test lets it, and both ends of the connection
 * have socket buffers far smaller than {@link #BLOCKING_MESSAGE_SIZE}, so that a test which
 * publishes a message of that size holds the session thread of the end under test inside its
 * write for as long as it wants - the state in which a message published behind it is queued
 * and not written.
 */
@SuppressWarnings("javadoc")
public class SessionTest extends ReplicationTestCase
{
  private static final int SOCKET_TIMEOUT_MS = 30000;
  /**
   * Socket buffers small enough that {@link #BLOCKING_MESSAGE_SIZE} bytes cannot be written
   * through them: the write blocks until the peer reads. Set explicitly on both ends, since the
   * buffers the kernel picks on its own grow well beyond it on a loopback link - and the message
   * is larger by far than what they hold, since a kernel which does not honour the size asked
   * for on the receiving side, as macOS does not, must still be unable to take the whole of it.
   */
  private static final int SOCKET_BUFFER_SIZE = 8 * 1024;
  private static final int BLOCKING_MESSAGE_SIZE = 4 * 1024 * 1024;
  /** Time given to a callback which must not run, to see that it does not. */
  private static final long SETTLE_MS = 500;
  private static final int SENDER_ID = 1;
  private static final int PEER_ID = 2;

  @Test
  public void theCallbackRunsOnceTheMessageIsWrittenAndNotWhenItIsQueued() throws Exception
  {
    try (SessionPair pair = connectSessionPair())
    {
      pair.publisher.start();
      pair.publisher.waitForStartup();

      // The session thread is inside the write of this message until the peer reads it.
      pair.publisher.publish(newBlockingMsg());
      pair.awaitBytesReachedThePeer();

      final CountDownLatch written = new CountDownLatch(1);
      final boolean accepted = pair.publisher.publish(new HeartbeatMsg(), written::countDown);

      assertThat(accepted).as("the message was refused by a running session").isTrue();
      assertThat(written.await(SETTLE_MS, TimeUnit.MILLISECONDS))
          .as("the callback ran while the message was still queued behind a message the peer "
              + "had not read")
          .isFalse();

      assertThat(pair.peer.receive()).isInstanceOf(EntryMsg.class);
      assertThat(written.await(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS))
          .as("the callback did not run once the message had been written")
          .isTrue();
      assertThat(pair.peer.receive()).isInstanceOf(HeartbeatMsg.class);
    }
  }

  @Test
  public void aMessageWhichCannotBeEncodedForThePeerIsRefusedWithoutRunningTheCallback()
      throws Exception
  {
    try (SessionPair pair = connectSessionPair())
    {
      pair.publisher.start();
      pair.publisher.waitForStartup();
      // A ReplicaOfflineMsg has no encoding before protocol version 8.
      pair.publisher.setProtocolVersion(ProtocolVersion.REPLICATION_PROTOCOL_V7);

      final CountDownLatch written = new CountDownLatch(1);
      final boolean accepted =
          pair.publisher.publish(new ReplicaOfflineMsg(newCSN()), written::countDown);

      assertThat(accepted)
          .as("a message the peer cannot decode was reported as accepted")
          .isFalse();
      assertThat(written.await(SETTLE_MS, TimeUnit.MILLISECONDS))
          .as("the callback ran for a message which was never written")
          .isFalse();
    }
  }

  /**
   * Before its thread is started, and once that thread is gone, a session writes on the
   * publishing thread itself; the callback then runs on that same thread, after the write.
   */
  @Test
  public void theCallbackRunsAfterAMessageWrittenOnThePublishingThread() throws Exception
  {
    try (SessionPair pair = connectSessionPair())
    {
      final CountDownLatch written = new CountDownLatch(1);
      final boolean accepted = pair.publisher.publish(new HeartbeatMsg(), written::countDown);

      assertThat(accepted).as("the message was refused by a session with no thread").isTrue();
      assertThat(written.getCount())
          .as("the callback had not run when the publish which wrote the message returned")
          .isZero();
      assertThat(pair.peer.receive()).isInstanceOf(HeartbeatMsg.class);
    }
  }

  private static EntryMsg newBlockingMsg()
  {
    return new EntryMsg(SENDER_ID, PEER_ID, new byte[BLOCKING_MESSAGE_SIZE], 1);
  }

  private static CSN newCSN()
  {
    return new CSNGenerator(SENDER_ID, 0).newCSN();
  }

  /**
   * Connects the end under test, in the server role of the replication protocol, with a peer
   * which reads only when a test does. Both ends exchange one message under TLS and then drop the
   * security layer, as the replication handshake does when encryption is not required, so that
   * the socket buffers alone decide when a write blocks: a message read under TLS by each end is
   * what consumes the records TLS itself sends after its negotiation, which would otherwise be
   * read as the start of a replication message once the layer is gone.
   */
  private static SessionPair connectSessionPair() throws Exception
  {
    final ReplSessionSecurity security = getReplSessionSecurity();
    final ExecutorService executor = Executors.newSingleThreadExecutor();
    final Socket peerSocket = new Socket();
    Socket publisherSocket = null;
    Session publisher = null;
    boolean connected = false;
    try (ServerSocket listen = TestCaseUtils.bindFreePort())
    {
      listen.setSoTimeout(SOCKET_TIMEOUT_MS);
      peerSocket.setReceiveBufferSize(SOCKET_BUFFER_SIZE);
      peerSocket.setTcpNoDelay(true);
      peerSocket.connect(new InetSocketAddress("127.0.0.1", listen.getLocalPort()), SOCKET_TIMEOUT_MS);
      // The TLS negotiation needs both ends handshaking at the same time.
      final Future<Session> peerEnd =
          executor.submit(() -> security.createClientSession(peerSocket, SOCKET_TIMEOUT_MS));

      publisherSocket = listen.accept();
      publisherSocket.setSendBufferSize(SOCKET_BUFFER_SIZE);
      publisherSocket.setTcpNoDelay(true);
      publisher = security.createServerSession(publisherSocket, SOCKET_TIMEOUT_MS);
      assertThat(publisher).as("could not create the session under test").isNotNull();
      final Session peer = peerEnd.get(SOCKET_TIMEOUT_MS, TimeUnit.MILLISECONDS);

      publisher.publish(new HeartbeatMsg());
      assertThat(peer.receive()).isInstanceOf(HeartbeatMsg.class);
      peer.publish(new HeartbeatMsg());
      assertThat(publisher.receive()).isInstanceOf(HeartbeatMsg.class);
      publisher.stopEncryption();
      peer.stopEncryption();
      connected = true;
      return new SessionPair(publisher, peer, peerSocket);
    }
    finally
    {
      executor.shutdownNow();
      if (!connected)
      {
        if (publisher != null)
        {
          publisher.close();
        }
        StaticUtils.close(publisherSocket, peerSocket);
      }
    }
  }

  private static final class SessionPair implements Closeable
  {
    private final Session publisher;
    private final Session peer;
    private final Socket peerSocket;

    private SessionPair(Session publisher, Session peer, Socket peerSocket)
    {
      this.publisher = publisher;
      this.peer = peer;
      this.peerSocket = peerSocket;
    }

    /**
     * Waits for the first bytes of a message to reach the peer: the session thread of the end
     * under test is then inside the write of that message, and stays there until the peer
     * reads, since the message is larger than the buffers on both sides of the connection.
     */
    void awaitBytesReachedThePeer() throws Exception
    {
      final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(SOCKET_TIMEOUT_MS);
      while (peerSocket.getInputStream().available() == 0)
      {
        assertThat(System.nanoTime() < deadline)
            .as("nothing was written to the peer")
            .isTrue();
        Thread.sleep(10);
      }
    }

    @Override
    public void close()
    {
      // The peer first: a session thread held inside a write is released by the peer going away,
      // and close() joins that thread.
      peer.close();
      publisher.close();
    }
  }
}