joystream/storage-node/packages/colossus/lib/sync.js

/*
 * This file is part of the storage node for the Joystream project.
 * Copyright (C) 2019 Joystream Contributors
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */

'use strict'

const debug = require('debug')('joystream:sync')

async function sync_callback(api, storage) {
	// The first step is to gather all data objects from chain.
	// TODO: in future, limit to a configured tranche
	// FIXME this isn't actually on chain yet, so we'll fake it.
	const knownContentIds = (await api.assets.getKnownContentIds()) || []

	const role_addr = api.identities.key.address
	const providerId = api.storageProviderId

	// Iterate over all sync objects, and ensure they're synced.
	const allChecks = knownContentIds.map(async (content_id) => {
		let { relationship, relationshipId } = await api.assets.getStorageRelationshipAndId(providerId, content_id)

		// get the data object
		// make sure the data object was Accepted by the liaison,
		// don't just blindly attempt to fetch them

		let fileLocal
		try {
			// check if we have content or not
			const stats = await storage.stat(content_id)
			fileLocal = stats.local
		} catch (err) {
			// on error stating or timeout
			debug(err.message)
			// we don't have content if we can't stat it
			fileLocal = false
		}

		if (!fileLocal) {
			try {
				await storage.synchronize(content_id)
			} catch (err) {
				// duplicate logging
				// debug(err.message)
				return
			}
			// why are we returning, if we synced the file
			return
		}

		if (!relationship) {
			// create relationship
			debug(`Creating new storage relationship for ${content_id.encode()}`)
			try {
				relationshipId = await api.assets.createAndReturnStorageRelationship(role_addr, providerId, content_id)
				await api.assets.toggleStorageRelationshipReady(role_addr, providerId, relationshipId, true)
			} catch (err) {
				debug(`Error creating new storage relationship ${content_id.encode()}: ${err.stack}`)
				return
			}
		} else if (!relationship.ready) {
			debug(`Updating storage relationship to ready for ${content_id.encode()}`)
			// update to ready. (Why would there be a relationship set to ready: false?)
			try {
				await api.assets.toggleStorageRelationshipReady(role_addr, providerId, relationshipId, true)
			} catch (err) {
				debug(`Error setting relationship ready ${content_id.encode()}: ${err.stack}`)
			}
		} else {
			// we already have content and a ready relationship set. No need to do anything
			// debug(`content already stored locally ${content_id.encode()}`);
		}
	})

	return Promise.all(allChecks)
}

async function sync_periodic(api, flags, storage) {
	try {
		debug('Starting sync run...')
		await sync_callback(api, storage)
		debug('sync run complete')
	} catch (err) {
		debug(`Error in sync_periodic ${err.stack}`)
	}
	// always try again
	setTimeout(sync_periodic, flags.syncPeriod, api, flags, storage)
}

function start_syncing(api, flags, storage) {
	sync_periodic(api, flags, storage)
}

module.exports = {
	start_syncing,
}