/*
 * 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 2023-2026 3A Systems, LLC.
 */
package org.opends.server.backends.cassandra;


import static org.opends.server.backends.pluggable.spi.StorageUtils.addErrorMessage;
import static org.opends.server.util.StaticUtils.stackTraceToSingleLineString;

import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.Set;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.forgerock.opendj.config.server.ConfigChangeResult;
import org.forgerock.opendj.config.server.ConfigException;
import org.forgerock.opendj.config.server.ConfigurationChangeListener;
import org.forgerock.opendj.ldap.ByteSequence;
import org.forgerock.opendj.ldap.ByteString;
import org.forgerock.opendj.server.config.server.CASBackendCfg;
import org.opends.server.backends.pluggable.spi.AccessMode;
import org.opends.server.backends.pluggable.spi.Cursor;
import org.opends.server.backends.pluggable.spi.Importer;
import org.opends.server.backends.pluggable.spi.ReadOnlyStorageException;
import org.opends.server.backends.pluggable.spi.ReadOperation;
import org.opends.server.backends.pluggable.spi.ReadableTransaction;
import org.opends.server.backends.pluggable.spi.SequentialCursor;
import org.opends.server.backends.pluggable.spi.StorageRuntimeException;
import org.opends.server.backends.pluggable.spi.StorageStatus;
import org.opends.server.backends.pluggable.spi.TreeName;
import org.opends.server.backends.pluggable.spi.UpdateFunction;
import org.opends.server.backends.pluggable.spi.WriteOperation;
import org.opends.server.backends.pluggable.spi.WriteableTransaction;
import org.opends.server.core.ServerContext;
import org.opends.server.types.BackupConfig;
import org.opends.server.types.BackupDirectory;
import org.opends.server.types.DirectoryException;
import org.opends.server.types.RestoreConfig;
import org.opends.server.util.BackupManager;

import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
import com.datastax.oss.driver.api.core.cql.BoundStatement;
import com.datastax.oss.driver.api.core.cql.PreparedStatement;
import com.datastax.oss.driver.api.core.cql.ResultSet;
import com.datastax.oss.driver.api.core.cql.Row;
import com.datastax.oss.driver.api.core.cql.Statement;

import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.LoadingCache;

public class CASStorage implements org.opends.server.backends.pluggable.spi.Storage, ConfigurationChangeListener<CASBackendCfg>{
	
	private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

	private CASBackendCfg config;

	//a cursor starts with a small page so that a point lookup does not transfer thousands of rows,
	//and doubles it while a scan keeps outrunning it (the driver default page is 5000 rows)
	final int initialPageSize=Math.max(1,Integer.getInteger("org.openidentityplatform.opendj.cassandra.fetchsize.initial",32));
	final int maxPageSize=Math.max(initialPageSize,Integer.getInteger("org.openidentityplatform.opendj.cassandra.fetchsize",1000));

	//CursorImpl.forwardInPage outcomes
	final static int POSITIONED=1;
	final static int MISSING=0;
	final static int NEEDS_SEEK=-1;
	final static int PAGE_OUT=-2;

	public CASStorage(CASBackendCfg cfg, ServerContext serverContext) {
		this.config = cfg;
		cfg.addCASChangeListener(this);
	}

	//config
	@Override
	public boolean isConfigurationChangeAcceptable(CASBackendCfg configuration,List<LocalizableMessage> unacceptableReasons) {
		return true;
	}

	@Override
	public ConfigChangeResult applyConfigurationChange(CASBackendCfg cfg) {
		final ConfigChangeResult ccr = new ConfigChangeResult();
		try
		{
			this.config = cfg;
		}
		catch (Exception e)
		{
			addErrorMessage(ccr, LocalizableMessage.raw(stackTraceToSingleLineString(e)));
		}
		return ccr;
	}

	CqlSession session=null;

	final LoadingCache<String,PreparedStatement> prepared = Caffeine.newBuilder()
		.expireAfterAccess(Duration.ofMinutes(10))
		.maximumSize(4096)
		.build(query -> session.prepare(query));

	ResultSet execute(Statement<?> statement) {
		if (logger.isTraceEnabled()) {
			final ResultSet res=session.execute(statement.setTracing(true));
			logger.trace(LocalizableMessage.raw(
					"cassandra: %s"
					,res.getExecutionInfo().getQueryTrace().getParameters()
				)
			);
			return res;
		}
		return session.execute(statement);
	}
	
	AccessMode accessMode=null;
	@Override
	public void open(AccessMode accessMode) throws Exception {
		this.accessMode=accessMode;
		session=CqlSession.builder()
			.withApplicationName("OpenDJ "+getKeyspaceName()+"."+config.getBackendId())
			.withConfigLoader(DriverConfigLoader.fromDefaults(CASStorage.class.getClassLoader()))
			.build();
		if (AccessMode.READ_WRITE.equals(accessMode)) {
			execute(prepared.get("CREATE KEYSPACE IF NOT EXISTS "+getKeyspaceName()+" WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'};").bind().setExecutionProfileName(profile));
		}
		storageStatus = StorageStatus.working();
	}

	private StorageStatus storageStatus = StorageStatus.lockedDown(LocalizableMessage.raw("closed"));
	@Override
	public StorageStatus getStorageStatus() {
		return storageStatus;
	}
	
	@Override
	public void close() {
		storageStatus = StorageStatus.lockedDown(LocalizableMessage.raw("closed"));
		if (session!=null && !session.isClosed()) {
			session.close();
		}
		session=null;
	}

	String getKeyspaceName() {
		return "\""+System.getProperty("keyspace",config.getDBDirectory()).replaceAll("[^a-zA-Z0-9_]", "_")+"\"";
	}
	
	String getTableName() {
		return getKeyspaceName()+".\""+config.getBackendId().replaceAll("[^a-zA-Z0-9_]", "_")+"\"";
	}
	
	@Override
	public void removeStorageFiles() throws StorageRuntimeException {
		final boolean isOpen=getStorageStatus().isWorking();
		if (!isOpen) {
			try {
				open(AccessMode.READ_WRITE);
			}catch (Exception e) {
				throw new StorageRuntimeException(e);
			}
		}
		try {
			execute(prepared.get("TRUNCATE TABLE "+getTableName()+";").bind().setExecutionProfileName(profile));
		}catch (Throwable e) {}
		if (!isOpen) {
			close();
		}
	}
	
	//operation
	@Override
	public <T> T read(ReadOperation<T> readOperation) throws Exception {
		return readOperation.run(new TransactionImpl(AccessMode.READ_ONLY));
	}

	@Override
	public void write(WriteOperation writeOperation) throws Exception {
		writeOperation.run(new TransactionImpl(accessMode));
	}

	final static String profile="ddl";
	static {
		if (System.getProperty("datastax-java-driver.basic.request.timeout")==null) {
			System.setProperty("datastax-java-driver.basic.request.timeout", "10 seconds");
		}
		if (System.getProperty("datastax-java-driver.profiles."+profile+".basic.request.timeout")==null) {
			System.setProperty("datastax-java-driver.profiles."+profile+".basic.request.timeout", "30 seconds");
		}
	}
	private final class TransactionImpl implements ReadableTransaction,WriteableTransaction {
		
		final AccessMode accessMode;
		public TransactionImpl(AccessMode accessMode) {
			super();
			this.accessMode=accessMode;
		}

		@Override
		public void openTree(TreeName name, boolean createOnDemand) {
			if (createOnDemand) {
				execute(prepared.get("CREATE TABLE IF NOT EXISTS "+getTableName()+" (baseDN text,indexId text,key blob,value blob,PRIMARY KEY ((baseDN,indexId),key));").bind().setExecutionProfileName(profile));
			}
		}
		
		public void clearTree(TreeName treeName) {
			checkReadOnly();
			deleteTree(treeName);
		}
		
		@Override
		public ByteString read(TreeName treeName, ByteSequence key) {
			final Row row=execute(
				prepared.get("SELECT value FROM "+getTableName()+" WHERE baseDN=:baseDN and indexId=:indexId and key=:key").bind()
					.setString("baseDN", treeName.getBaseDN()).setString("indexId", treeName.getIndexId())
					.setByteBuffer("key", ByteBuffer.wrap(key.toByteArray()))
			).one();
			return row==null?null:ByteString.wrap(row.getByteBuffer("value").array());
		}

		@Override
		public Cursor<ByteString, ByteString> openCursor(TreeName treeName) {
			return new CursorImpl(this,treeName);
		}

		@Override
		public long getRecordCount(TreeName treeName) {
			return execute(
				prepared.get("SELECT count(*) FROM "+getTableName()+" WHERE baseDN=:baseDN and indexId=:indexId").bind()
					.setString("baseDN", treeName.getBaseDN()).setString("indexId", treeName.getIndexId())
			).one().getLong(0);
		}

		@Override
		public void deleteTree(TreeName treeName) {
			checkReadOnly();
			openTree(treeName,true);
			execute(
				prepared.get("DELETE FROM "+getTableName()+" WHERE baseDN=:baseDN and indexId=:indexId").bind()
					.setString("baseDN", treeName.getBaseDN()).setString("indexId", treeName.getIndexId())
			);
		}

		@Override
		public void put(TreeName treeName, ByteSequence key, ByteSequence value) {
			checkReadOnly();
			execute(
				prepared.get("INSERT INTO "+getTableName()+" (baseDN,indexId,key,value) VALUES (:baseDN,:indexId,:key,:value)").bind()
					.setString("baseDN", treeName.getBaseDN()).setString("indexId", treeName.getIndexId())
					.setByteBuffer("key", ByteBuffer.wrap(key.toByteArray()))
					.setByteBuffer("value",ByteBuffer.wrap(value.toByteArray()))
			);
		}

		@Override
		public boolean update(TreeName treeName, ByteSequence key, UpdateFunction f) {
			checkReadOnly();
			final ByteString oldValue=read(treeName,key);
			final ByteSequence newValue=f.computeNewValue(oldValue);
			if (Objects.equals(newValue, oldValue))
			{
				return false;
			}
			if (newValue == null)
			{
				delete(treeName, key);
				return true;
			}
			put(treeName,key,newValue);
			return true;
		}

		@Override
		public boolean delete(TreeName treeName, ByteSequence key) {
			checkReadOnly();
			execute(
				prepared.get("DELETE FROM "+getTableName()+" WHERE baseDN=:baseDN and indexId=:indexId and key=:key").bind()
					.setString("baseDN", treeName.getBaseDN()).setString("indexId", treeName.getIndexId())
					.setByteBuffer("key", ByteBuffer.wrap(key.toByteArray()))
			);
			return true;
		}
		
		void checkReadOnly() {
			if (AccessMode.READ_ONLY.equals(accessMode)) {
				throw new ReadOnlyStorageException();
			}
		}
	}
	
	// Iterates the (baseDN,indexId) partition in key order.
	// A ResultSet can only be consumed once: rc.iterator() always returns the same iterator,
	// so "restarting" it never rewinds. Every repositioning that cannot be served by moving
	// forward within the already-fetched rows therefore runs a new server-side slice query
	// on the "key" clustering column instead.
	// Pages are sized by the cursor rather than by the driver default of 5000 rows: a point
	// lookup must not transfer thousands of entries, so a cursor starts with a small page and
	// doubles it (up to maxPageSize) every time a scan outruns it. Scanning past the end of a
	// page runs the next slice from the current key instead of letting the driver page with the
	// size the cursor started with.
	final class CursorImpl implements Cursor<ByteString, ByteString> {
		final TreeName treeName;
		final TransactionImpl tx;
		final String tableName=getTableName();

		//visible for tests
		long queryCount;
		int pageSize=initialPageSize;

		ResultSet rc;
		Iterator<Row> iterator;
		//the row the cursor is on, or - when defined is false - the row a failed positionToKey
		//stopped just before, which is where next() resumes (same as pdb)
		Row current=null;
		boolean defined=false;
		//rc holds a DESC page: it must never serve a forward move
		boolean descending=false;
		boolean closed=false;

		public CursorImpl(TransactionImpl tx,TreeName treeName) {
			this.treeName=treeName;
			this.tx=tx;
			//lazy: the first navigation decides which query to run, a seek must not pay for a partition scan
		}

		ResultSet select(String condition,ByteSequence key,int rows){
			queryCount++;
			BoundStatement statement=prepared.get("SELECT key,value FROM "+tableName+" WHERE baseDN=:baseDN and indexId=:indexId"+condition).bind()
				.setString("baseDN", treeName.getBaseDN()).setString("indexId", treeName.getIndexId())
				.setPageSize(rows);
			if (key!=null) {
				statement=statement.setByteBuffer("key", ByteBuffer.wrap(key.toByteArray()));
			}
			return execute(statement);
		}

		//runs a new page and positions the cursor on its first row
		boolean slice(String condition,ByteSequence key){
			rc=select(condition,key,pageSize);
			iterator=rc.iterator();
			descending=false;
			if (iterator.hasNext()) {
				current=iterator.next();
				defined=true;
				return true;
			}
			current=null;
			defined=false;
			return false;
		}

		boolean seek(ByteSequence key){
			return slice(" and key>=:key ORDER BY key",key);
		}

		void growPage() {
			pageSize=(int)Math.min(maxPageSize,2L*pageSize);
		}

		//serves a forward repositioning from the rows the driver already fetched: they are the
		//sorted rows following the current one (blob clustering collates in unsigned byte order)
		int forwardInPage(ByteSequence key,boolean exactMatch) {
			if (!defined || descending || rc==null) {
				return NEEDS_SEEK;
			}
			int cmp=key.compareTo(getKey());
			if (cmp==0) {
				return POSITIONED;
			}
			if (cmp<0) { //backward: only the server can rewind
				return NEEDS_SEEK;
			}
			while (rc.getAvailableWithoutFetching()>0) {
				current=iterator.next();
				cmp=key.compareTo(getKey());
				if (cmp==0) {
					return POSITIONED;
				}
				if (cmp<0) { //walked past the key
					if (exactMatch) {
						defined=false; //the cursor stops just before this row
						return MISSING;
					}
					return POSITIONED;
				}
			}
			return PAGE_OUT;
		}

		@Override
		public boolean next() {
			if (closed) {
				return false;
			}
			if (current!=null && !defined) { //a failed positionToKey stopped just before this row
				defined=true;
				return true;
			}
			if (rc==null) { //lazy cursor: the first navigation runs the scan
				return slice(" ORDER BY key",null);
			}
			if (!descending && rc.getAvailableWithoutFetching()>0) {
				current=iterator.next();
				defined=true;
				return true;
			}
			if (current==null) { //exhausted or explicitly undefined: stay there
				return false;
			}
			if (!descending && rc.isFullyFetched()) { //the server has nothing left either
				current=null;
				defined=false;
				return false;
			}
			//page boundary: continue with a bigger slice starting right after the current key
			growPage();
			return slice(" and key>:key ORDER BY key",getKey());
		}

		@Override
		public boolean isDefined() {
			return defined;
		}

		@Override
		public ByteString getKey() throws NoSuchElementException {
			if (!isDefined()) {
				throw new NoSuchElementException();
			}
			return ByteString.wrap(current.getByteBuffer("key").array());
		}

		@Override
		public ByteString getValue() throws NoSuchElementException {
			if (!isDefined()) {
				throw new NoSuchElementException();
			}
			return ByteString.wrap(current.getByteBuffer("value").array());
		}

		@Override
		public void delete() throws NoSuchElementException, UnsupportedOperationException {
			if (!isDefined()) {
				throw new NoSuchElementException();
			}
			tx.delete(treeName, getKey());
		}

		//a closed cursor is undefined and every navigation on it returns false, like EmptyCursor
		@Override
		public void close() {
			closed=true;
			iterator=Collections.emptyIterator();
			current=null;
			defined=false;
			descending=false;
			rc=null;
		}


		@Override
		public boolean positionToKeyOrNext(ByteSequence key) {
			if (closed) {
				return false;
			}
			final int served=forwardInPage(key,false);
			if (served>=0) {
				return served==POSITIONED;
			}
			if (served==PAGE_OUT) { //the scan outran its page: the next slice should be bigger
				growPage();
			}
			return seek(key);
		}

		@Override
		public boolean positionToKey(ByteSequence key) {
			if (closed) {
				return false;
			}
			final int served=forwardInPage(key,true);
			if (served>=0) {
				return served==POSITIONED;
			}
			if (served==PAGE_OUT) {
				growPage();
			}
			if (seek(key) && key.compareTo(getKey())==0) {
				return true;
			}
			//like jeb/pdb a miss leaves the cursor undefined; the row the seek landed on is the
			//first key after the missing one, so next() resumes there instead of skipping it
			defined=false;
			return false;
		}


		@Override
		public boolean positionToLastKey() {
			if (closed) {
				return false;
			}
			rc=select(" ORDER BY key DESC LIMIT 1",null,1);
			iterator=rc.iterator();
			descending=true;
			if (iterator.hasNext()) {
				current=iterator.next(); //nothing follows the last key
				defined=true;
				return true;
			}
			current=null;
			defined=false;
			return false;
		}

		@Override
		public boolean positionToIndex(int index) {
			if (closed) {
				return false;
			}
			if (index<0) {
				rc=null; //an invalid index resets the cursor: next() starts the scan again
				iterator=null;
				current=null;
				defined=false;
				descending=false;
				return false;
			}
			//CQL has no offset clause: restart from the first row and skip. The rows that have to
			//be walked are asked for in one page instead of one round trip per page
			rc=select(" ORDER BY key",null,(int)Math.min(maxPageSize,Math.max(pageSize,index+1L)));
			iterator=rc.iterator();
			descending=false;
			for (int ct=0;ct<=index;ct++) {
				if (!iterator.hasNext()) {
					current=null;
					defined=false;
					return false;
				}
				current=iterator.next();
			}
			defined=true;
			return true;
		}
	}
	
	@Override
	public Set<TreeName> listTrees() {
		// TODO Auto-generated method stub
		return Collections.emptySet();
	}
	

	private final class ImporterImpl implements Importer {
		final TransactionImpl tx;
		
		final Boolean isOpen;
		
		public ImporterImpl() {
			isOpen=getStorageStatus().isWorking();
			if (!isOpen) {
				try {
					open(AccessMode.READ_WRITE);
				}catch (Exception e) {
					throw new StorageRuntimeException(e);
				}
			}
			tx=new TransactionImpl(accessMode);
		}
		
		@Override
		public void close() {
			if (!isOpen) {
				CASStorage.this.close();
			}
		}
		
		@Override
		public void clearTree(TreeName name) {
			tx.clearTree(name);
		}
		
		@Override
		public void put(TreeName treeName, ByteSequence key, ByteSequence value) {
			tx.put(treeName, key, value);
		}
		
		@Override
		public ByteString read(TreeName treeName, ByteSequence key) {
			return tx.read(treeName, key);
		}
		
		@Override
		public SequentialCursor<ByteString, ByteString> openCursor(TreeName treeName) {
			return tx.openCursor(treeName);
		}
	}
	
	//import
	@Override
	public Importer startImport() throws ConfigException, StorageRuntimeException {
		return new ImporterImpl();
	}
	
	//backup
	@Override
	public boolean supportsBackupAndRestore() {
		return true;
	}

	@Override
	public void createBackup(BackupConfig backupConfig) throws DirectoryException
	{
		// TODO backup over snapshot or cassandra export
		//new BackupManager(config.getBackendId()).createBackup(this, backupConfig);
	}

	@Override
	public void removeBackup(BackupDirectory backupDirectory, String backupID) throws DirectoryException
	{
		new BackupManager(config.getBackendId()).removeBackup(backupDirectory, backupID);
	}

	@Override
	public void restoreBackup(RestoreConfig restoreConfig) throws DirectoryException
	{
		// TODO restore over snapshot or cassandra export
		//new BackupManager(config.getBackendId()).restoreBackup(this, restoreConfig);
	}

}