/*
 * Copyright 1999-2023 Alibaba Group Holding Ltd.
 *
 * Licensed 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 com.alibaba.nacos.core.persistence;

import com.alibaba.nacos.common.notify.NotifyCenter;
import com.alibaba.nacos.consistency.snapshot.LocalFileMeta;
import com.alibaba.nacos.consistency.snapshot.Reader;
import com.alibaba.nacos.consistency.snapshot.SnapshotOperation;
import com.alibaba.nacos.consistency.snapshot.Writer;
import com.alibaba.nacos.persistence.constants.PersistenceConstant;
import com.alibaba.nacos.persistence.datasource.DataSourceService;
import com.alibaba.nacos.persistence.datasource.DynamicDataSource;
import com.alibaba.nacos.persistence.datasource.LocalDataSourceServiceImpl;
import com.alibaba.nacos.persistence.model.event.DerbyLoadEvent;
import com.alibaba.nacos.persistence.utils.PersistenceExecutor;
import com.alibaba.nacos.sys.env.EnvUtil;
import com.alibaba.nacos.sys.utils.DiskUtils;
import com.alibaba.nacos.sys.utils.TimerContext;
import com.alipay.sofa.jraft.util.CRC64;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.io.File;
import java.nio.file.Paths;
import java.sql.CallableStatement;
import java.sql.Connection;
import java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.BiConsumer;
import java.util.zip.Checksum;

/**
 * Derby Snapshot operation.
 * TODO depend on jraft strongly, Waiting for addition split.
 *
 * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
 */
public class DerbySnapshotOperation implements SnapshotOperation {
    
    private static final Logger LOGGER = LoggerFactory.getLogger(DerbySnapshotOperation.class);
    
    private static final String DERBY_SNAPSHOT_SAVE = DerbySnapshotOperation.class.getSimpleName() + ".SAVE";
    
    private static final String DERBY_SNAPSHOT_LOAD = DerbySnapshotOperation.class.getSimpleName() + ".LOAD";
    
    private final String backupSql = "CALL SYSCS_UTIL.SYSCS_BACKUP_DATABASE(?)";
    
    private final String snapshotDir = "derby_data";
    
    private final String snapshotArchive = "derby_data.zip";
    
    private final String derbyBaseDir = Paths.get(EnvUtil.getNacosHome(), "data", PersistenceConstant.DERBY_BASE_DIR)
            .toString();
    
    private final String restoreDB = "jdbc:derby:" + derbyBaseDir;
    
    private final String checkSumKey = "checkSum";
    
    private final ReentrantReadWriteLock.WriteLock writeLock;
    
    public DerbySnapshotOperation(ReentrantReadWriteLock.WriteLock writeLock) {
        this.writeLock = writeLock;
    }
    
    @Override
    public void onSnapshotSave(Writer writer, BiConsumer<Boolean, Throwable> callFinally) {
        PersistenceExecutor.executeSnapshot(() -> {
            TimerContext.start(DERBY_SNAPSHOT_SAVE);
            
            final Lock lock = writeLock;
            lock.lock();
            try {
                final String writePath = writer.getPath();
                final String parentPath = Paths.get(writePath, snapshotDir).toString();
                DiskUtils.deleteDirectory(parentPath);
                DiskUtils.forceMkdir(parentPath);
                
                doDerbyBackup(parentPath);
                
                final String outputFile = Paths.get(writePath, snapshotArchive).toString();
                final Checksum checksum = new CRC64();
                DiskUtils.compress(writePath, snapshotDir, outputFile, checksum);
                DiskUtils.deleteDirectory(parentPath);
                
                final LocalFileMeta meta = new LocalFileMeta();
                meta.append(checkSumKey, Long.toHexString(checksum.getValue()));
                
                callFinally.accept(writer.addFile(snapshotArchive, meta), null);
            } catch (Throwable t) {
                LOGGER.error("Fail to compress snapshot, path={}, file list={}, {}.", writer.getPath(),
                        writer.listFiles(), t);
                callFinally.accept(false, t);
            } finally {
                lock.unlock();
                TimerContext.end(DERBY_SNAPSHOT_SAVE, LOGGER);
            }
        });
    }
    
    @Override
    public boolean onSnapshotLoad(Reader reader) {
        final String readerPath = reader.getPath();
        final String sourceFile = Paths.get(readerPath, snapshotArchive).toString();
        TimerContext.start(DERBY_SNAPSHOT_LOAD);
        final Lock lock = writeLock;
        lock.lock();
        try {
            final Checksum checksum = new CRC64();
            DiskUtils.decompress(sourceFile, readerPath, checksum);
            
            LocalFileMeta fileMeta = reader.getFileMeta(snapshotArchive);
            
            if (fileMeta.getFileMeta().containsKey(checkSumKey)) {
                if (!Objects.equals(Long.toHexString(checksum.getValue()), fileMeta.get(checkSumKey))) {
                    throw new IllegalArgumentException("Snapshot checksum failed");
                }
            }
            
            final String loadPath = Paths.get(readerPath, snapshotDir, PersistenceConstant.DERBY_BASE_DIR).toString();
            LOGGER.info("snapshot load from : {}, and copy to : {}", loadPath, derbyBaseDir);
            
            doDerbyRestoreFromBackup(() -> {
                final File srcDir = new File(loadPath);
                final File destDir = new File(derbyBaseDir);
                
                DiskUtils.copyDirectory(srcDir, destDir);
                LOGGER.info("Complete database recovery");
                return null;
            });
            DiskUtils.deleteDirectory(loadPath);
            NotifyCenter.publishEvent(DerbyLoadEvent.INSTANCE);
            return true;
        } catch (final Throwable t) {
            LOGGER.error("Fail to load snapshot, path={}, file list={}, {}.", readerPath, reader.listFiles(), t);
            return false;
        } finally {
            lock.unlock();
            TimerContext.end(DERBY_SNAPSHOT_LOAD, LOGGER);
        }
    }
    
    private void doDerbyBackup(String backupDirectory) throws Exception {
        DataSourceService sourceService = DynamicDataSource.getInstance().getDataSource();
        DataSource dataSource = sourceService.getJdbcTemplate().getDataSource();
        try (Connection holder = Objects.requireNonNull(dataSource, "dataSource").getConnection()) {
            CallableStatement cs = holder.prepareCall(backupSql);
            cs.setString(1, backupDirectory);
            cs.execute();
        }
    }
    
    private void doDerbyRestoreFromBackup(Callable<Void> callable) throws Exception {
        DataSourceService sourceService = DynamicDataSource.getInstance().getDataSource();
        LocalDataSourceServiceImpl localDataSourceService = (LocalDataSourceServiceImpl) sourceService;
        localDataSourceService.restoreDerby(restoreDB, callable);
    }
    
}
