joplin/packages/lib/services/ResourceFetcher.ts

import Resource from '../models/Resource';
import Setting from '../models/Setting';
import BaseService from './BaseService';
import ResourceService from './ResourceService';
import Logger from '../Logger';
import shim from '../shim';
const { Dirnames } = require('./synchronizer/utils/types');
const EventEmitter = require('events');

export default class ResourceFetcher extends BaseService {

	public static instance_: ResourceFetcher;

	public dispatch: Function = (_o: any) => {};
	private logger_: Logger = new Logger();
	private queue_: any[] = [];
	private fetchingItems_: any = {};
	private maxDownloads_ = 3;
	private addingResources_ = false;
	private eventEmitter_ = new EventEmitter();
	private autoAddResourcesCalls_: any[] = [];
	private fileApi_: any;
	private updateReportIID_: any;
	private scheduleQueueProcessIID_: any;
	private scheduleAutoAddResourcesIID_: any;

	constructor(fileApi: any = null) {
		super();
		this.setFileApi(fileApi);
	}

	static instance() {
		if (ResourceFetcher.instance_) return ResourceFetcher.instance_;
		ResourceFetcher.instance_ = new ResourceFetcher();
		return ResourceFetcher.instance_;
	}

	on(eventName: string, callback: Function) {
		return this.eventEmitter_.on(eventName, callback);
	}

	off(eventName: string, callback: Function) {
		return this.eventEmitter_.removeListener(eventName, callback);
	}

	setLogger(logger: Logger) {
		this.logger_ = logger;
	}

	logger() {
		return this.logger_;
	}

	setFileApi(v: any) {
		if (v !== null && typeof v !== 'function') throw new Error(`fileApi must be a function that returns the API. Type is ${typeof v}`);
		this.fileApi_ = v;
	}

	async fileApi() {
		return this.fileApi_();
	}

	queuedItemIndex_(resourceId: string) {
		for (let i = 0; i < this.fetchingItems_.length; i++) {
			const item = this.fetchingItems_[i];
			if (item.id === resourceId) return i;
		}
		return -1;
	}

	updateReport() {
		const fetchingCount = Object.keys(this.fetchingItems_).length;
		this.dispatch({
			type: 'RESOURCE_FETCHER_SET',
			fetchingCount: fetchingCount,
			toFetchCount: fetchingCount + this.queue_.length,
		});
	}

	async markForDownload(resourceIds: string[]) {
		if (!Array.isArray(resourceIds)) resourceIds = [resourceIds];

		const fetchStatuses = await Resource.fetchStatuses(resourceIds);

		const idsToKeep = [];
		for (const status of fetchStatuses) {
			if (status.fetch_status !== Resource.FETCH_STATUS_IDLE) continue;
			idsToKeep.push(status.resource_id);
		}

		for (const id of idsToKeep) {
			await Resource.markForDownload(id);
		}

		for (const id of idsToKeep) {
			this.queueDownload_(id, 'high');
		}
	}

	queueDownload_(resourceId: string, priority: string = null) {
		if (priority === null) priority = 'normal';

		const index = this.queuedItemIndex_(resourceId);
		if (index >= 0) return false;
		if (this.fetchingItems_[resourceId]) return false;

		const item = { id: resourceId };

		if (priority === 'high') {
			this.queue_.splice(0, 0, item);
		} else {
			this.queue_.push(item);
		}

		this.updateReport();

		this.scheduleQueueProcess();
		return true;
	}

	async startDownload_(resourceId: string) {
		if (this.fetchingItems_[resourceId]) return;
		this.fetchingItems_[resourceId] = true;

		this.updateReport();

		const resource = await Resource.load(resourceId);
		const localState = await Resource.localState(resource);

		const completeDownload = async (emitDownloadComplete = true, localResourceContentPath = '') => {
			// 2019-05-12: This is only necessary to set the file size of the resources that come via
			// sync. The other ones have been done using migrations/20.js. This code can be removed
			// after a few months.
			if (resource && resource.size < 0 && localResourceContentPath && !resource.encryption_blob_encrypted) {
				await shim.fsDriver().waitTillExists(localResourceContentPath);
				await ResourceService.autoSetFileSizes();
			}

			delete this.fetchingItems_[resource.id];
			this.logger().debug(`ResourceFetcher: Removed from fetchingItems: ${resource.id}. New: ${JSON.stringify(this.fetchingItems_)}`);
			this.scheduleQueueProcess();

			// Note: This downloadComplete event is not really right or useful because the resource
			// might still be encrypted and the caller usually can't do much with this. In particular
			// the note being displayed will refresh the resource images but since they are still
			// encrypted it's not useful. Probably, the views should listen to DecryptionWorker events instead.
			if (resource && emitDownloadComplete) this.eventEmitter_.emit('downloadComplete', { id: resource.id, encrypted: !!resource.encryption_blob_encrypted });
			this.updateReport();
		};

		if (!resource) {
			this.logger().info(`ResourceFetcher: Attempting to download a resource that does not exist (has been deleted?): ${resourceId}`);
			await completeDownload(false);
			return;
		}

		// Shouldn't happen, but just to be safe don't re-download the
		// resource if it's already been downloaded.
		if (localState.fetch_status === Resource.FETCH_STATUS_DONE) {
			await completeDownload(false);
			return;
		}

		const fileApi = await this.fileApi();

		if (!fileApi) {
			this.logger().debug('ResourceFetcher: Disabled because fileApi is not set');
			return;
		}

		this.fetchingItems_[resourceId] = resource;

		const localResourceContentPath = Resource.fullPath(resource, !!resource.encryption_blob_encrypted);
		const remoteResourceContentPath = `${Dirnames.Resources}/${resource.id}`;

		await Resource.setLocalState(resource, { fetch_status: Resource.FETCH_STATUS_STARTED });

		this.logger().debug(`ResourceFetcher: Downloading resource: ${resource.id}`);

		this.eventEmitter_.emit('downloadStarted', { id: resource.id });

		fileApi
			.get(remoteResourceContentPath, { path: localResourceContentPath, target: 'file' })
		// eslint-disable-next-line promise/prefer-await-to-then -- Old code before rule was applied
			.then(async () => {
				await Resource.setLocalState(resource, { fetch_status: Resource.FETCH_STATUS_DONE });
				this.logger().debug(`ResourceFetcher: Resource downloaded: ${resource.id}`);
				await completeDownload(true, localResourceContentPath);
			})
		// eslint-disable-next-line promise/prefer-await-to-then -- Old code before rule was applied
			.catch(async (error: any) => {
				this.logger().error(`ResourceFetcher: Could not download resource: ${resource.id}`, error);
				await Resource.setLocalState(resource, { fetch_status: Resource.FETCH_STATUS_ERROR, fetch_error: error.message });
				await completeDownload();
			});
	}

	processQueue_() {
		while (Object.getOwnPropertyNames(this.fetchingItems_).length < this.maxDownloads_) {
			if (!this.queue_.length) break;
			const item = this.queue_.splice(0, 1)[0];
			void this.startDownload_(item.id);
		}

		if (!this.queue_.length) {
			void this.autoAddResources(10);
		}
	}

	async waitForAllFinished() {
		return new Promise((resolve) => {
			const iid = shim.setInterval(() => {
				if (!this.updateReportIID_ &&
                    !this.scheduleQueueProcessIID_ &&
                    !this.queue_.length &&
                    !this.autoAddResourcesCalls_.length &&
                    !Object.getOwnPropertyNames(this.fetchingItems_).length) {

					shim.clearInterval(iid);
					resolve(null);
				}
			}, 100);
		});
	}

	async autoAddResources(limit: number = null) {
		this.autoAddResourcesCalls_.push(true);
		try {
			if (limit === null) limit = 10;

			if (this.addingResources_) return;
			this.addingResources_ = true;

			this.logger().info(`ResourceFetcher: Auto-add resources: Mode: ${Setting.value('sync.resourceDownloadMode')}`);

			let count = 0;
			const resources = await Resource.needToBeFetched(Setting.value('sync.resourceDownloadMode'), limit);
			for (let i = 0; i < resources.length; i++) {
				const added = this.queueDownload_(resources[i].id);
				if (added) count++;
			}

			this.logger().info(`ResourceFetcher: Auto-added resources: ${count}`);

			const errorCount = await Resource.downloadStatusCounts(Resource.FETCH_STATUS_ERROR);
			if (errorCount) this.dispatch({ type: 'SYNC_HAS_DISABLED_SYNC_ITEMS' });

		} finally {
			this.addingResources_ = false;
			this.autoAddResourcesCalls_.pop();
		}
	}

	async start() {
		await Resource.resetStartedFetchStatus();
		void this.autoAddResources(10);
	}

	scheduleQueueProcess() {
		if (this.scheduleQueueProcessIID_) {
			shim.clearTimeout(this.scheduleQueueProcessIID_);
			this.scheduleQueueProcessIID_ = null;
		}

		this.scheduleQueueProcessIID_ = shim.setTimeout(() => {
			this.processQueue_();
			this.scheduleQueueProcessIID_ = null;
		}, 100);
	}

	scheduleAutoAddResources() {
		if (this.scheduleAutoAddResourcesIID_) return;

		this.scheduleAutoAddResourcesIID_ = shim.setTimeout(() => {
			this.scheduleAutoAddResourcesIID_ = null;
			void ResourceFetcher.instance().autoAddResources();
		}, 1000);
	}

	async fetchAll() {
		await Resource.resetStartedFetchStatus();
		void this.autoAddResources(null);
	}

	async destroy() {
		this.eventEmitter_.removeAllListeners();
		if (this.scheduleQueueProcessIID_) {
			shim.clearTimeout(this.scheduleQueueProcessIID_);
			this.scheduleQueueProcessIID_ = null;
		}
		if (this.scheduleAutoAddResourcesIID_) {
			shim.clearTimeout(this.scheduleAutoAddResourcesIID_);
			this.scheduleAutoAddResourcesIID_ = null;
		}
		await this.waitForAllFinished();
		this.eventEmitter_ = null;
		ResourceFetcher.instance_ = null;
	}
}