% FileStore File storage shared by MATLAB clients and workers.
% FileStore is file storage owned by a specific job. Each
% entry consists of a file and its corresponding key. When the
% owning job is deleted, the store is deleted as well.
%
% Use the FileStore to store files from workers that can be retrieved by 
% clients later, even while the job is still running.
%
% MATLAB clients and workers are free to read and modify the FileStore at
% any time. Files added and removed by MATLAB clients will be visible to
% other MATLAB clients and workers even while the job is running. Files
% added and removed by MATLAB workers will also be visible to other MATLAB
% clients and workers even while the job is running. The ordering of
% operations executed by different processes is not guaranteed.
%
% Note: For parallel.cluster.Generic clusters with 'HasSharedFileSystem'
% set to false, the visibility of modifications made to the FileStore while a job
% is running will depend on your specific implementation.
% Without additional synchronization between the client JobStorageLocation
% and the worker JobStorageLocation, changes may only be visible once the
% job has completed.
%
% Examples:
%
%   To copy a file from the local file system to the FileStore:
%       copyFileToStore(myFileStore, "/tmp/local.data", "myKey");
%   The entry key in the FileStore will be "myKey".
%   To copy multiple files from the local file system to the FileStore:
%       copyFileToStore(myFileStore, ["/tmp/local.data", "/data/local.txt"], ...
%           ["myKey1", "myKey2"]);
%
%   To copy an entry from the FileStore to the local file system:
%       copyFileFromStore(myFileStore, "myKey", "/data/myNewName.data");
%   This will copy the entry to "/data/myNewName.data" on the local file system.
%   To copy multiple entries from the FileStore to the local file system:
%       copyFileFromStore(myFileStore, ["myKey1", "myKey2"], ...
%           ["/tmp/new.data", "/data/new.txt"]);
%
%   To delete an existing entry in the FileStore:
%       remove(myFileStore, "myKey");
%   To delete multiple existing entries in the FileStore:
%       remove(myFileStore, ["myKey1", "myKey2"]);
%
% FileStore methods:
%   isKey             -   Determine whether FileStore contains given keys
%   keys              -   Return string array of all keys of FileStore
%   copyFileToStore   -   Copy files from local file system to FileStore
%   copyFileFromStore -   Copy entries from FileStore to local file system
%   remove            -   Remove entries from FileStore
%
% FileStore properties:
%   KeyUpdatedFcn  -   Callback executed when an entry is added or replaced.
%   KeyRemovedFcn  -   Callback executed when an entry is removed.
%
% See Also getCurrentFileStore, parallel.ValueStore
 
% Copyright 2021-2023 The MathWorks, Inc.

classdef (ConstructOnLoad = true) FileStore < parallel.internal.keyvalue.Store
    
    methods (Static, Hidden)
        function store = createDummyForDisplay()
            store = parallel.FileStore([], '', false, false, false, false);
        end

        function store = hCreateFileBased(path)
            impl = parallel.internal.keyvalue.CJSKeyValueStore(path);
            store = parallel.FileStore(impl, '', true, true, true, true);
        end
    end

    methods (Hidden, Access = {?parallel.internal.cluster.CJSJobMixin, ...
            ?parallel.internal.cluster.MJSJobMixin,...
            ?parallel.internal.pool.ThreadsKeyValueStoreAdapter})
        function obj = FileStore(impl, owner, ...
                workersSeeClientUpdates, workersSeeClientRemoves, ...
                clientsSeeWorkerUpdates, clientsSeeWorkerRemoves)
            parallel.FileStore.loadCheck(nargin);
            obj = obj@parallel.internal.keyvalue.Store(impl, ...
                workersSeeClientUpdates, workersSeeClientRemoves, ...
                clientsSeeWorkerUpdates, clientsSeeWorkerRemoves);
            % Invalidate this store when the parent job is destroyed
            if ~isempty(owner)
                addlistener(owner , 'ObjectBeingDestroyed', @(~, ~) delete(obj));
            end
        end
    end

    methods
        function tf = isKey(obj, keys)
            %ISKEY Determine whether FileStore contains given keys
            %   TF = ISKEY(FS, KEYS) returns a logical indicating whether the
            %   FileStore object, FS, contains the keys, KEYS. TF is a
            %   N element logical array, where N is the number of elements in 
            %   KEYS.

            validateattributes(obj, {'parallel.FileStore'}, {'scalar'}, 'isKey', 'obj');
            validateattributes(keys, {'char', 'string', 'cell'}, {}, 'isKey', 'keys');
            keys = convertCharsToStrings(keys);
            tf = obj.doIsKey(keys);
        end

        function allKeys = keys(obj)
            %KEYS Return a string array of all keys in FileStore
            %   ALLKEYS = KEYS(FS) returns a N x 1 string array of all the keys
            %   in the FileStore object, FS.

            validateattributes(obj, {'parallel.FileStore'}, {'scalar'}, 'keys', 'obj');
            allKeys = obj.doKeys();
        end

        function copyFileToStore(obj, localFiles, keys)
            %COPYFILETOSTORE Copy files from local file system to FileStore
            %   COPYFILETOSTORE(FS, LOCALFILES, KEYS) copies files from
            %   the local file system to the FileStore object, FS.
            %   LOCALFILES is a string array of N local file paths. KEYS 
            %   is a string array of N keys. Each file in LOCALFILES
            %   is copied to the corresponding FileStore entry in KEYS.

            validateattributes(obj, {'parallel.FileStore'}, {'scalar'}, 'copyFileToStore', 'obj');
            validateattributes(localFiles, {'char', 'string', 'cell'}, {}, 'copyFileToStore', 'localFiles');
            validateattributes(keys, {'char', 'string', 'cell'}, {}, 'copyFileToStore', 'keys');
            [localFiles, keys] = convertCharsToStrings(localFiles, keys);
            iValidateVectorisedInputSizes(keys, localFiles);
            if isempty(localFiles)
                return
            end

            try
                arrayfun(@iVerifyFileExists, localFiles);
                obj.Impl.putFile(localFiles, keys);
            catch E
                err = MException(message('parallel:job:StoreWriteError', strjoin(keys, ', ')));
                err = err.addCause(E);
                throw(err);
            end
        end

        function copyFileFromStore(obj, keys, localFiles)
            %COPYFILEFROMSTORE Copy files from FileStore to local file system
            %   COPYFILEFROMSTORE(FS, KEYS, LOCALFILES) copies files from
            %   the FileStore object, FS, to the local file system.
            %   KEYS is a string array of N keys. LOCALFILES is a 
            %   string array of N local file paths. Each entry in KEYS is
            %   copied to its corresponding destination in LOCALFILES.
            
            validateattributes(obj, {'parallel.FileStore'}, {'scalar'}, 'copyFileFromStore', 'obj');
            validateattributes(keys, {'char', 'string', 'cell'}, {}, 'copyFileFromStore', 'keys');
            validateattributes(localFiles, {'char', 'string', 'cell'}, {}, 'copyFileFromStore', 'localFiles');
            [keys, localFiles] = convertCharsToStrings(keys, localFiles);
            iValidateVectorisedInputSizes(keys, localFiles);
            if isempty(localFiles)
                return
            end

            % Use a temp staging area when copying multiple files. This is
            % to ensure we get either all or none of the files.
            useStagingArea = ~isscalar(localFiles);
            try
                localFiles = arrayfun(@iValidateDestination, keys, localFiles);
                if useStagingArea
                    stagingFolder = tempname;
                    parallel.internal.files.FileUtils.mkdir(stagingFolder);
                    destroyStagingFolder = onCleanup(@() parallel.internal.files.FileUtils.safeDeleteFolder(stagingFolder));
                    stagingDestinations = cellfun(@(~) string(tempname(stagingFolder)), localFiles);
                    obj.Impl.getFile(keys, stagingDestinations);

                    % Successfully wrote all staging files. Move them into user
                    % specified destination
                    cellfun(@(s, l) iMoveToFinalDestination(s, l), stagingDestinations, localFiles);
                else
                    obj.Impl.getFile(keys, localFiles);
                end
            catch E
                err = MException(message('parallel:job:StoreReadError', strjoin(keys, ', ')));
                err = err.addCause(E);
                throw(err);
            end
        end

        function remove(obj, keys)
            %REMOVE Remove entries from FileStore
            %   REMOVE(FS, KEYS) removes one or more entries from the
            %   FileStore object, FS. KEYS is a string array of N keys
            %   to remove from the FileStore.

            validateattributes(obj, {'parallel.FileStore'}, {'scalar'}, 'remove', 'obj');
            validateattributes(keys, {'char', 'string', 'cell'}, {}, 'remove', 'keys');
            keys = convertCharsToStrings(keys);
            if isempty(keys)
                return
            end
            
            obj.Impl.remove(keys);
       end
    end

    methods (Access = private, Static)
        function loadCheck(ctorNargin)
        %loadCheck if the constructor is being called with zero arguments,
        %assumption is that the store is being loaded. Therefore error.
            if ctorNargin == 0
                error(message('parallel:job:CannotLoadFileStoreFromMatFile'));
            end
        end
    end
end

function iVerifyFileExists(file)
    if isfolder(file)
        error(message('parallel:job:StoreFolderSourceNotSupported', file));
    end
    if ~isfile(file)
        error(message('parallel:job:StoreFileNotFound', file));
    end
end

function destination = iValidateDestination(key, destination)
    if isfolder(destination)
        % Destination is an existing folder. Generate file name under the 
        % given directory using the key
        destination = fullfile(destination, key);
    else
        parentFolder = fileparts(destination);
        if parentFolder ~= "" && ~isfolder(parentFolder)
            % Parent to destination is not an existing folder. This will fail
            % since we cannot autocreate the parent folder structure
            error(message('parallel:job:StoreDestinationDoesNotExist', destination));
        end
    end
end

function iMoveToFinalDestination(tempSource, destination)
% moveFile with error handling to give the same error as if we did not use
% a staging location
    try
        parallel.internal.files.FileUtils.moveFile(tempSource, destination);
    catch E
        if ~isempty(E.cause)
            cause = E.cause{1};
            if strcmp(cause.identifier, 'MATLAB:MOVEFILE:ResourceNotFound')
                error(message('parallel:job:StoreDestinationDoesNotExist', destination));
            end
        end
        rethrow(E);
    end
end

function iValidateVectorisedInputSizes(in1, in2)
    if numel(in2) ~= numel(in1)
        error(message('parallel:job:StoreNumberOfVariablesMismatch'));
    end
end
