import assert from 'node:assert';
import async from 'async';
import backupSites from '../backupsites.js';
import BoxError from '../boxerror.js';
import DataLayout from '../datalayout.js';
import logger from '../logger.js';
import hush from '../hush.js';
const { DecryptStream, EncryptStream } = hush;
import fs from 'node:fs';
import HashStream from '../hash-stream.js';
import path from 'node:path';
import paths from '../paths.js';
import { pipeline } from 'node:stream/promises';
import ProgressStream from '../progress-stream.js';
import promiseRetry from '../promise-retry.js';
import safe from 'safetydance';
import shellModule from '../shell.js';
import syncer from '../syncer.js';
import util from 'node:util';
import { Writable } from 'node:stream';

const { log } = logger('backupformat/rsync');
const shell = shellModule('backupformat/rsync');

async function addFile(sourceFile, encryption, uploader, progressCallback) {
    assert.strictEqual(typeof sourceFile, 'string');
    assert.strictEqual(typeof encryption, 'object');
    assert.strictEqual(typeof uploader, 'object');
    assert.strictEqual(typeof progressCallback, 'function');

    // make sure file can be opened for reading before we start the pipeline. otherwise, we end up with
    // destinations dirs/file which are owned by root (this process id) and cannot be copied (run as normal user)
    const [openError, sourceHandle] = await safe(fs.promises.open(sourceFile, 'r'));
    if (openError) {
        log(`addFile: ignoring disappeared file: ${sourceFile}`);
        return { integrity: null, stats: { transferred: 0 } };
    }

    const sourceStream = sourceHandle.createReadStream(sourceFile, { autoClose: true });
    const ps = new ProgressStream({ interval: 10000 }); // display a progress every 10 seconds
    ps.on('progress', function (progress) {
        const transferred = Math.round(progress.transferred/1024/1024), speed = Math.round(progress.speed/1024/1024);
        if (!transferred && !speed) return progressCallback({ message: `Uploading ${sourceFile}` }); // 0M@0MBps looks wrong
        progressCallback({ message: `Uploading ${sourceFile}: ${transferred}M@${speed}MBps` });
    });
    ps.on('heartbeat', function ({ elapsed, transferred }) {
        progressCallback({ message: `Still uploading ${sourceFile} (${elapsed}s, ${Math.round(transferred/1024/1024)}M)` });
    });

    // careful not to have async code between here and pipeline() for 'error' handling
    const hash = new HashStream();
    const destStream = uploader.createStream();

    let pipelinePromise;
    if (encryption) {
        const encryptStream = new EncryptStream(encryption);
        pipelinePromise = pipeline(sourceStream, encryptStream, ps, hash, destStream);
    } else {
        pipelinePromise = pipeline(sourceStream, ps, hash, destStream);
    }

    const [error] = await safe(pipelinePromise);
    if (error && !error.message.includes('ENOENT')) throw new BoxError(BoxError.EXTERNAL_ERROR, `tarPack pipeline error for ${sourceFile}: ${error.message}`); // ignore error if file disappears
    // log(`addFile: pipeline finished: ${JSON.stringify(ps.stats())}`);

    await uploader.finish();

    return {
        stats: { transferred: ps.stats().transferred },
        integrity: { size: ps.stats().transferred, sha256: hash.digest('hex') }
    };
}

async function saveFsMetadata(dataLayout, metadataFile) {
    assert(dataLayout instanceof DataLayout, 'dataLayout must be a DataLayout');
    assert.strictEqual(typeof metadataFile, 'string');

    // contains paths prefixed with './'
    const metadata = {
        emptyDirs: [],
        execFiles: [],
        symlinks: []
    };

    // we assume small number of files. spawnSync will raise a ENOBUFS error after maxBuffer
    for (const lp of dataLayout.localPaths()) {
        const [emptyDirsError, emptyDirs] = await safe(shell.spawn('find', [lp, '-type', 'd', '-empty'], { encoding: 'utf8', maxLines: 80000 }));
        if (emptyDirsError && emptyDirsError.stdoutLineCount >= 80000) throw new BoxError(BoxError.FS_ERROR, `Too many empty directories. Run "find ${lp} -type d -empty" to investigate`);
        if (emptyDirsError) throw emptyDirsError;
        if (emptyDirs.length) metadata.emptyDirs = metadata.emptyDirs.concat(emptyDirs.trim().split('\n').map((ed) => dataLayout.toRemotePath(ed)));

        const [execFilesError, execFiles] = await safe(shell.spawn('find', [lp, '-type', 'f', '-executable'], { encoding: 'utf8', maxLines: 20000 }));
        if (execFilesError && execFilesError.stdoutLineCount >= 20000) throw new BoxError(BoxError.FS_ERROR, `Too many executable files. Run "find ${lp} -type f -executable" to investigate`);
        if (execFilesError) throw execFilesError;
        if (execFiles.length) metadata.execFiles = metadata.execFiles.concat(execFiles.trim().split('\n').map((ef) => dataLayout.toRemotePath(ef)));

        const [symlinkFilesError, symlinkFiles] = await safe(shell.spawn('find', [lp, '-type', 'l'], { encoding: 'utf8', maxLines: 20000 }));
        if (symlinkFilesError && symlinkFilesError.stdoutLineCount >= 20000) throw new BoxError(BoxError.FS_ERROR, `Too many symlinks. Run "find ${lp} -type l" to investigate`);
        if (symlinkFilesError) throw symlinkFilesError;

        if (symlinkFiles.length) metadata.symlinks = metadata.symlinks.concat(symlinkFiles.trim().split('\n').map((sl) => {
            const site = safe.fs.readlinkSync(sl);
            return { path: dataLayout.toRemotePath(sl), site };
        }));
    }

    if (!safe.fs.writeFileSync(metadataFile, JSON.stringify(metadata, null, 2))) throw new BoxError(BoxError.FS_ERROR, `Error writing fs metadata: ${safe.error.message}`);
}

// this is not part of 'snapshotting' because we need root access to traverse
async function restoreFsMetadata(dataLayout, metadataFile) {
    assert(dataLayout instanceof DataLayout, 'dataLayout must be a DataLayout');
    assert.strictEqual(typeof metadataFile, 'string');

    log(`Recreating empty directories in ${dataLayout.toString()}`);

    const metadataJson = safe.fs.readFileSync(metadataFile, 'utf8');
    if (metadataJson === null) throw new BoxError(BoxError.EXTERNAL_ERROR, 'Error loading fsmetadata.json:' + safe.error.message);
    const metadata = safe.JSON.parse(metadataJson);
    if (metadata === null) throw new BoxError(BoxError.EXTERNAL_ERROR, 'Error parsing fsmetadata.json:' + safe.error.message);

    for (const emptyDir of metadata.emptyDirs) {
        const [mkdirError] = await safe(fs.promises.mkdir(dataLayout.toLocalPath(emptyDir), { recursive: true }));
        if (mkdirError) throw new BoxError(BoxError.FS_ERROR, `unable to create path: ${mkdirError.message}`);
    }

    for (const execFile of metadata.execFiles) {
        const [chmodError] = await safe(fs.promises.chmod(dataLayout.toLocalPath(execFile), parseInt('0755', 8)));
        if (chmodError) throw new BoxError(BoxError.FS_ERROR, `unable to chmod: ${chmodError.message}`);
    }

    for (const symlink of (metadata.symlinks || [])) {
        if (!symlink.site) continue;
        // the path may not exist if we had a directory full of symlinks
        const [mkdirError] = await safe(fs.promises.mkdir(path.dirname(dataLayout.toLocalPath(symlink.path)), { recursive: true }));
        if (mkdirError) throw new BoxError(BoxError.FS_ERROR, `unable to symlink (mkdir): ${mkdirError.message}`);
        const [symlinkError] = await safe(fs.promises.symlink(symlink.site, dataLayout.toLocalPath(symlink.path), 'file'));
        if (symlinkError) throw new BoxError(BoxError.FS_ERROR, `unable to symlink: ${symlinkError.message}`);
    }
}

async function copy(backupSite, fromPath, toPath, progressCallback) {
    assert.strictEqual(typeof backupSite, 'object');
    assert.strictEqual(typeof fromPath, 'string');
    assert.strictEqual(typeof toPath, 'string');
    assert.strictEqual(typeof progressCallback, 'function');

    await backupSites.storageApi(backupSite).copyDir(backupSite.config, backupSite.limits, fromPath, toPath, progressCallback);
}

function getFileExtension(encryption) {
    assert.strictEqual(typeof encryption, 'boolean');

    return ''; // this also signals to backupcleanear that we are dealing with directories
}

async function sync(backupSite, remotePath, dataLayout, progressCallback) {
    assert.strictEqual(typeof backupSite, 'object');
    assert.strictEqual(typeof remotePath, 'string');
    assert(dataLayout instanceof DataLayout, 'dataLayout must be a DataLayout');
    assert.strictEqual(typeof progressCallback, 'function');

    // the number here has to take into account the s3.upload partSize (which is 10MB). So 20=200MB
    const concurrency = backupSite.limits?.syncConcurrency || (backupSite.provider === 's3' ? 20 : 10);
    const cacheFile = path.join(paths.BACKUP_INFO_DIR, backupSite.id, `${dataLayout.getBasename()}.sync.cache`);
    const { delQueue, addQueue, integrityMap } = await syncer.sync(dataLayout, cacheFile); // integrityMap is unchanged files
    log(`sync: processing ${delQueue.length} deletes, ${addQueue.length} additions and ${integrityMap.size} unchanged`);
    const aggregatedStats = {
        transferred: 0,
        size: [...integrityMap.values()].reduce((sum, integrity) => sum + (integrity?.size || 0), 0), // integrity can be null if file had disappeared during upload
        fileCount: addQueue.length + integrityMap.size, // final file count, not the transferred file count
    };

    const destPathIntegrityMap = new Map(); // unlike integrityMap which contains local filenames, this contains destination filenames (maybe encrypted)
    for (const [entryPath, integrity] of integrityMap) {
        const destPath = backupSite.encryption?.encryptedFilenames ? hush.encryptFilePath(entryPath, backupSite.encryption) : entryPath;
        destPathIntegrityMap.set(destPath, integrity);
    }

    let completedAdds = 0;
    function reportUploadProgress(detail) {
        const summary = `Uploaded ${completedAdds}/${addQueue.length} files`;
        progressCallback({ message: detail ? `${summary}. ${detail}` : summary });
    }

    async function processSyncerChange(change) {
        // the empty task.path is special to signify the directory
        const destPath = change.path && backupSite.encryption?.encryptedFilenames ? hush.encryptFilePath(change.path, backupSite.encryption) : change.path;
        const fullPath = path.join(remotePath, destPath);

        if (change.operation === 'removedir') {
            await backupSites.storageApi(backupSite).removeDir(backupSite.config, backupSite.limits, fullPath, progressCallback);
        } else if (change.operation === 'remove') {
            await backupSites.storageApi(backupSite).remove(backupSite.config, fullPath);
        } else if (change.operation === 'add') {
            await promiseRetry({ times: 5, interval: 20000, log }, async (retryCount) => {
                reportUploadProgress(`Current: ${change.path}` + (retryCount > 1 ? ` (Try ${retryCount})` : ''));
                if (retryCount > 1) log(`sync: retrying ${change.path} position ${change.position} try ${retryCount}`);
                const uploader = await backupSites.storageApi(backupSite).upload(backupSite.config, backupSite.limits, fullPath);
                const result = await addFile(dataLayout.toLocalPath('./' + change.path), backupSite.encryption, uploader, (progress) => {
                    reportUploadProgress(progress.message);
                });
                integrityMap.set(change.path, result.integrity); // .integrity can be null when file disappeared on us
                destPathIntegrityMap.set(destPath, result.integrity);
                aggregatedStats.transferred += result.stats.transferred;
                aggregatedStats.size += result.stats.transferred;
            });
            ++completedAdds;
        }
    }

    const [delError] = await safe(async.eachLimit(delQueue, concurrency, async (change) => await processSyncerChange(change, backupSite, remotePath, dataLayout, progressCallback)));
    log('sync: done processing deletes. error: %o', delError);
    if (delError) throw delError;

    const [addError] = await safe(async.eachLimit(addQueue, concurrency, async (change) => await processSyncerChange(change, backupSite, remotePath, dataLayout, progressCallback)));
    log('sync: done processing adds. error: %o', addError);
    if (addError) throw addError;

    progressCallback({ message: `Uploaded ${completedAdds} files` });

    await syncer.finalize(integrityMap, cacheFile);

    return {
        stats: aggregatedStats,
        integrityMap: destPathIntegrityMap
    };
}

async function downloadDir(backupSite, remotePath, dataLayout, progressCallback) {
    assert.strictEqual(typeof backupSite, 'object');
    assert.strictEqual(typeof remotePath, 'string');
    assert(dataLayout instanceof DataLayout, 'dataLayout must be a DataLayout');
    assert.strictEqual(typeof progressCallback, 'function');

    const encryptedFilenames = backupSite.encryption?.encryptedFilenames || false;

    log(`downloadDir: ${remotePath} to ${dataLayout.toString()}. encryption filenames: ${encryptedFilenames}. encrypted files: ${!!backupSite.encryption}`);

    let completedFiles = 0, totalFiles = 0;
    let lastProgressTime = 0;
    function reportProgress(detail) {
        const now = Date.now();
        if (now - lastProgressTime < 5000) return;
        lastProgressTime = now;
        const summary = `Downloaded ${completedFiles}/${totalFiles} files`;
        progressCallback({ message: detail ? `${summary}. ${detail}` : summary });
    }

    async function downloadFile(entry) {
        let relativePath = path.relative(remotePath, entry.path);
        if (encryptedFilenames) {
            const { error, result } = hush.decryptFilePath(relativePath, backupSite.encryption);
            if (error) throw new BoxError(BoxError.CRYPTO_ERROR, `Unable to decrypt file ${entry.path}`);
            relativePath = result;
        }
        const destFilePath = dataLayout.toLocalPath('./' + relativePath);

        const [mkdirError] = await safe(fs.promises.mkdir(path.dirname(destFilePath), { recursive: true }));
        if (mkdirError) throw new BoxError(BoxError.FS_ERROR, mkdirError.message);

        await promiseRetry({ times: 3, interval: 20000 }, async function () {
            reportProgress(`Current: ${relativePath}`);

            const [downloadError, sourceStream] = await safe(backupSites.storageApi(backupSite).download(backupSite.config, entry.path));
            if (downloadError) {
                log(`downloadDir: download ${entry.path} to ${destFilePath} errored: ${downloadError.message}`);
                throw downloadError;
            }

            const ps = new ProgressStream({ interval: 10000 }); // display a progress every 10 seconds
            ps.on('progress', function (progress) {
                const transferred = Math.round(progress.transferred/1024/1024), speed = Math.round(progress.speed/1024/1024);
                if (transferred || speed) reportProgress(`Current: ${relativePath} ${transferred}M@${speed}MBps`);
            });
            ps.on('heartbeat', function ({ elapsed, transferred }) {
                progressCallback({ message: `Still downloading ${relativePath} (${elapsed}s, ${Math.round(transferred/1024/1024)}M)` });
            });

            const destStream = fs.createWriteStream(destFilePath);

            const streams = [ sourceStream, ps ];

            if (backupSite.encryption) {
                const decryptStream = new DecryptStream(backupSite.encryption);
                streams.push(decryptStream);
            }

            streams.push(destStream);

            const [pipelineError] = await safe(pipeline(streams));
            if (pipelineError) {
                log(`downloadDir: download error ${entry.path} to ${destFilePath}: ${pipelineError.message}`);
                throw pipelineError;
            }
        });

        ++completedFiles;
    }

    // https://www.digitalocean.com/community/questions/rate-limiting-on-spaces?answer=40441
    const concurrency = backupSite.limits?.downloadConcurrency || (backupSite.provider === 's3' ? 30 : 10);
    let marker = null;
    while (true) {
        const batch = await backupSites.storageApi(backupSite).listDir(backupSite.config, remotePath, marker === null ? 1 : 1000, marker); // try with one file first. if that works out, we continue faster
        totalFiles += batch.entries.length;
        await async.eachLimit(batch.entries, concurrency, downloadFile);
        if (!batch.marker) break;
        marker = batch.marker;
    }

    progressCallback({ message: `Downloaded ${completedFiles} files` });
}

async function download(backupSite, remotePath, dataLayout, progressCallback) {
    assert.strictEqual(typeof backupSite, 'object');
    assert.strictEqual(typeof remotePath, 'string');
    assert(dataLayout instanceof DataLayout, 'dataLayout must be a DataLayout');
    assert.strictEqual(typeof progressCallback, 'function');

    log(`download: Downloading ${remotePath} to ${dataLayout.toString()}`);

    await downloadDir(backupSite, remotePath, dataLayout, progressCallback);
    await restoreFsMetadata(dataLayout, `${dataLayout.localRoot()}/fsmetadata.json`);
}

async function upload(backupSite, remotePath, dataLayout, progressCallback) {
    assert.strictEqual(typeof backupSite, 'object');
    assert.strictEqual(typeof remotePath, 'string');
    assert.strictEqual(typeof dataLayout, 'object');
    assert.strictEqual(typeof progressCallback, 'function');

    log(`upload: uploading to site ${backupSite.id} path ${remotePath} (encrypted: ${!!backupSite.encryption}) dataLayout ${dataLayout.toString()}`);

    await saveFsMetadata(dataLayout, `${dataLayout.localRoot()}/fsmetadata.json`);
    return await sync(backupSite, remotePath, dataLayout, progressCallback); // { stats, integrityMap }
}

async function computeEntryIntegrity(backupSite, entry, progressCallback) {
    const sourceStream = await backupSites.storageApi(backupSite).download(backupSite.config, entry.path);

    const ps = new ProgressStream({ interval: 10000 }); // display a progress every 10 seconds
    ps.on('progress', function (progress) {
        const transferred = Math.round(progress.transferred/1024/1024), speed = Math.round(progress.speed/1024/1024);
        if (!transferred && !speed) return progressCallback({ message: `Downloading ${entry.path}` }); // 0M@0MBps looks wrong
        progressCallback({ message: `Downloading ${entry.path}: ${transferred}M@${speed}MBps` });
    });

    const streams = [ sourceStream, ps ];
    const hash = new HashStream();
    streams.push(hash);

    if (backupSite.encryption) {
        const decryptStream = new DecryptStream(backupSite.encryption);
        streams.push(decryptStream);
    }

    const sink = new Writable({ write(chunk, enc, cb) { cb(); } });
    streams.push(sink); // drain the stream

    const [error] = await safe(pipeline(streams));
    if (error) throw new BoxError(BoxError.EXTERNAL_ERROR, `pipeline error: ${error.message}`);

    return { transferred: ps.stats().transferred, digest: hash.digest() };
}

async function verify(backupSite, remotePath, integrityMap, progressCallback) {
    assert.strictEqual(typeof backupSite, 'object');
    assert.strictEqual(typeof remotePath, 'string');
    assert(util.types.isMap(integrityMap), 'integrityMap should be a Map');
    assert.strictEqual(typeof progressCallback, 'function');

    log(`verify: Verifying ${remotePath}`);

    // https://www.digitalocean.com/community/questions/rate-limiting-on-spaces?answer=40441
    const concurrency = backupSite.limits?.downloadConcurrency || (backupSite.provider === 's3' ? 30 : 10);
    let fileCount = 0, marker = null;
    const messages = [];
    while (true) {
        const batch = await backupSites.storageApi(backupSite).listDir(backupSite.config, remotePath, marker === null ? 1 : 1000, marker); // try with one file first. if that works out, we continue faster
        await async.eachLimit(batch.entries, concurrency, async function validateFile(entry) {
            const [error, result] = await safe(computeEntryIntegrity(backupSite, entry, progressCallback));
            if (error) return messages.push(`Could not verify integrity of ${entry.path}: ${error.message}`);

            const relativePath = path.relative(remotePath, entry.path);
            const integrity = integrityMap.get(relativePath);
            if (result.transferred !== integrity.size) {
                messages.push(`${entry.path} has size ${result.transferred}. Expecting ${integrity.size}`);
                log(`verify: size check of ${entry.path} failed: ${messages.at(-1)}`);
            } else if (result.digest !== integrity.sha256) {
                messages.push(`${entry.path} has digest ${result.digest}. Expecting ${integrity.sha256}`);
                log(`verify: digest check of ${entry.path} failed: ${messages.at(-1)}`);
            } else {
                log(`verify: ${entry.path} passed`);
            }
        });
        fileCount += batch.entries.length;
        if (!batch.marker) break;
        marker = batch.marker;
    }

    if (integrityMap.size !== fileCount) {
        messages.push(`Got ${fileCount} files. Expecting ${integrityMap.size} files`);
        log(`verify: file count mismatch: ${messages.at(-1)}`);
    }

    return messages;
}

export default {
    download,
    upload,
    verify,
    getFileExtension,
    copy,
    _saveFsMetadata: saveFsMetadata,
    _restoreFsMetadata: restoreFsMetadata,
};
