mirror of
https://github.com/fluencelabs/dweb-transports
synced 2025-04-24 14:22:17 +00:00
313 lines
11 KiB
JavaScript
313 lines
11 KiB
JavaScript
|
const Url = require('url');
|
|||
|
|
const Transport = require('./Transport'); // Base class for TransportXyz
|
|||
|
|
const Transports = require('./Transports'); // Manage all Transports that are loaded
|
|||
|
|
const errors = require('./Errors'); // Standard Dweb Errors
|
|||
|
|
const canonicaljson = require('@stratumn/canonicaljson');
|
|||
|
|
const fluence = require('fluence');
|
|||
|
|
|
|||
|
|
const debug = require('debug')('dweb-transports:fluence');
|
|||
|
|
|
|||
|
|
const defaultOptions = {
|
|||
|
|
nodeUrl: 'https://ia-redis.fluence.one',
|
|||
|
|
nodePort: 443,
|
|||
|
|
appId: '4' // Redis
|
|||
|
|
};
|
|||
|
|
|
|||
|
|
class TransportFluence extends Transport {
|
|||
|
|
|
|||
|
|
constructor(options) {
|
|||
|
|
super(options);
|
|||
|
|
this.options = options; // Dictionary of options
|
|||
|
|
this.session = undefined;
|
|||
|
|
this.name = 'FLUENCE'; // For console log etc
|
|||
|
|
this.supportURLs = ['fluence'];
|
|||
|
|
this.supportFunctions = [
|
|||
|
|
// General data functions
|
|||
|
|
'fetch', // p_rawfetch(url, {timeoutMS, start, end, relay}) – Fetch some bytes based on a url
|
|||
|
|
|
|||
|
|
// Lists functions
|
|||
|
|
'list', // p_rawlist(url) – Fetch all the objects in a list .. identified by the url of the .. 'signedby' parameter of the p_rawadd call
|
|||
|
|
'add', // p_rawadd(url, sig) – Store a new list item, it should be stored so that it can be retrieved either by "signedby" (using p_rawlist) or by "url" (with p_rawreverse).
|
|||
|
|
'newlisturls', // p_newlisturls(cl) – Obtain a pair of URLs for a new list
|
|||
|
|
|
|||
|
|
// KeyValueTable functions
|
|||
|
|
'newdatabase', // p_newdatabase(pubkey) – Create a new database based on some existing object
|
|||
|
|
'newtable', // p_newtable(pubkey, table) – Create a new table
|
|||
|
|
'get', // p_get(url, keys) – Get one or more keys from a table
|
|||
|
|
'set', // p_set(url, keyvalues, value) – Set one or more keys in a table.
|
|||
|
|
'getall',// p_getall(url) – Return a dictionary representing the table
|
|||
|
|
'keys', // p_keys(url) – Return a list of keys in a table (suitable for iterating through)
|
|||
|
|
];
|
|||
|
|
this.supportFeatures = [];
|
|||
|
|
this.status = Transport.STATUS_LOADED;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
static setup0(options) {
|
|||
|
|
const combinedOptions = Transport.mergeoptions(defaultOptions, options.fluence);
|
|||
|
|
|
|||
|
|
console.assert(combinedOptions.nodeUrl, 'Fluence Node url should be specified');
|
|||
|
|
console.assert(combinedOptions.nodePort, 'Fluence Node port should be specified');
|
|||
|
|
console.assert(combinedOptions.appId, 'Fluence AppId should be specified');
|
|||
|
|
|
|||
|
|
let t = new TransportFluence(combinedOptions);
|
|||
|
|
Transports.addtransport(t);
|
|||
|
|
return t;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async p_setup1(cb) {
|
|||
|
|
try {
|
|||
|
|
this.status = Transport.STATUS_STARTING;
|
|||
|
|
debug('connecting...');
|
|||
|
|
|
|||
|
|
if (cb) cb(this);
|
|||
|
|
|
|||
|
|
const rndString = Math.random().toString(36).substring(2, 15) + Math.random().toString(36).substring(2, 15);
|
|||
|
|
this.session = fluence.directConnect(this.options.nodeUrl, this.options.nodePort, this.options.appId, rndString);
|
|||
|
|
|
|||
|
|
debug('connected.');
|
|||
|
|
|
|||
|
|
this.status = Transport.STATUS_CONNECTED;
|
|||
|
|
} catch (err) {
|
|||
|
|
console.error(this.name, 'failed to start', err);
|
|||
|
|
this.status = Transport.STATUS_FAILED;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
if (cb) cb(this);
|
|||
|
|
return this;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
async p_status() {
|
|||
|
|
if (this.session !== null && this.session !== undefined) {
|
|||
|
|
this.status = Transport.STATUS_CONNECTED;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return super.p_status();
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
parseUrl(url) {
|
|||
|
|
const parsedUrl = Url.parse(url);
|
|||
|
|
if (parsedUrl.protocol !== 'fluence:') {
|
|||
|
|
throw new errors.TransportError(`TransportFLUENCE Error encountered retrieving val: url (${parsedUrl.href}) is not a valid FLUENCE url | protocol = ${parsedUrl.protocol}`);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
debug('retrieve url', parsedUrl.href);
|
|||
|
|
|
|||
|
|
return parsedUrl;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
newKey(pubkey) {
|
|||
|
|
if (pubkey.hasOwnProperty("keypair")) {
|
|||
|
|
pubkey = pubkey.keypair.signingexport();
|
|||
|
|
}
|
|||
|
|
// By this point pubkey should be an export of a public key of form xyz:abc where xyz
|
|||
|
|
// specifies the type of public key (NACL VERIFY being the only kind we expect currently)
|
|||
|
|
return `fluence:/fluence/${encodeURIComponent(pubkey)}`;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
parseRedisResponse(result) {
|
|||
|
|
|
|||
|
|
if (result.startsWith('$-1')) {
|
|||
|
|
return undefined;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
function parseResponsePart(result) {
|
|||
|
|
const [, type, countStr] = /^([\+\-\:\$\*]{1})([0-9]+)/.exec(result);
|
|||
|
|
const count = Number(countStr);
|
|||
|
|
|
|||
|
|
switch (type) {
|
|||
|
|
case '+': { // Simple string
|
|||
|
|
const offset = 1;
|
|||
|
|
const [, data] = /([^\n\r]+)/.exec(result.substring(offset));
|
|||
|
|
return {
|
|||
|
|
data: data,
|
|||
|
|
offset: offset + data.length + 2
|
|||
|
|
};
|
|||
|
|
}
|
|||
|
|
case ':': { // Integer
|
|||
|
|
const offset = 1;
|
|||
|
|
const [, data] = /([0-9]+)/.exec(result.substring(offset));
|
|||
|
|
return {
|
|||
|
|
data: Number(data),
|
|||
|
|
offset: offset + data.length + 2
|
|||
|
|
};
|
|||
|
|
}
|
|||
|
|
case '$': { // Bulk string
|
|||
|
|
const offset = 1 + String(count).length + 2;
|
|||
|
|
return {
|
|||
|
|
data: result.substring(offset, offset + count),
|
|||
|
|
offset: offset + count + 2
|
|||
|
|
};
|
|||
|
|
}
|
|||
|
|
case '*': { // Array
|
|||
|
|
let offset = 1 + String(count).length + 2;
|
|||
|
|
const list = [];
|
|||
|
|
for(let i = 0;i < count;i++) {
|
|||
|
|
const parsedListItem = parseResponsePart(result.substring(offset));
|
|||
|
|
list.push(parsedListItem.data);
|
|||
|
|
offset += parsedListItem.offset;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return {
|
|||
|
|
data: list,
|
|||
|
|
offset: null
|
|||
|
|
};
|
|||
|
|
}
|
|||
|
|
default: {
|
|||
|
|
throw new errors.TransportError(`TransportFLUENCE Error unsupprted Redis response type: ${type}, response: ${result}`);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return parseResponsePart(result).data;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// General data functions (uses Redis basic GET\SET)
|
|||
|
|
|
|||
|
|
async p_rawfetch(url) {
|
|||
|
|
const parsedUrl = this.parseUrl(url);
|
|||
|
|
const key = parsedUrl.path;
|
|||
|
|
|
|||
|
|
const result = await this.session.request(`GET ${key}`);
|
|||
|
|
const data = this.parseRedisResponse(result.asString());
|
|||
|
|
|
|||
|
|
if (!data) {
|
|||
|
|
throw new errors.TransportError(`TransportFLUENCE unable to retrieve: ${url.href}`);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return typeof data === 'string' ? JSON.parse(data) : data;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// List functions (uses Redis list)
|
|||
|
|
|
|||
|
|
async p_rawlist(url) {
|
|||
|
|
const parsedUrl = this.parseUrl(url);
|
|||
|
|
const key = parsedUrl.path;
|
|||
|
|
|
|||
|
|
const result = await this.session.request(`LRANGE ${key} 0 -1`);
|
|||
|
|
const data = this.parseRedisResponse(result.asString());
|
|||
|
|
|
|||
|
|
if (!data) {
|
|||
|
|
throw new errors.TransportError(`TransportFLUENCE unable to retrieve list: ${url.href}`);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return data.map(listItem => typeof listItem === 'string' ? JSON.parse(listItem) : listItem);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async p_rawadd(url, sig) {
|
|||
|
|
const parsedUrl = this.parseUrl(url);
|
|||
|
|
const key = parsedUrl.path;
|
|||
|
|
|
|||
|
|
const data = canonicaljson.stringify( sig.preflight( Object.assign({}, sig)));
|
|||
|
|
|
|||
|
|
await this.session.request(`RPUSH ${key} ${data}`);
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async p_newlisturls(cl) {
|
|||
|
|
const key = this.newKey(cl);
|
|||
|
|
|
|||
|
|
return [key, key];
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// KeyValueTable functions (uses Redis hashes)
|
|||
|
|
|
|||
|
|
async p_newdatabase(pubkey) {
|
|||
|
|
/*
|
|||
|
|
Request a new database
|
|||
|
|
returns: { publicurl: "fluence:/fluence/<publickey>", privateurl: "fluence:/fluence/<publickey>"> }
|
|||
|
|
*/
|
|||
|
|
let key = await this.newKey(pubkey);
|
|||
|
|
return {
|
|||
|
|
publicurl: key,
|
|||
|
|
privateurl: key
|
|||
|
|
};
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async p_newtable(pubkey, table) {
|
|||
|
|
/*
|
|||
|
|
Request a new table
|
|||
|
|
returns: {publicurl: "fluence:/fluence/<publickey>/<table>", privateurl: "fluence:/fluence/<publickey>/<table>">
|
|||
|
|
*/
|
|||
|
|
if (!pubkey) {
|
|||
|
|
throw new errors.CodingError("p_newtable currently requires a pubkey");
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
const { publicurl, privateurl } = await this.p_newdatabase(pubkey);
|
|||
|
|
return {
|
|||
|
|
privateurl: `${privateurl}/${table}`,
|
|||
|
|
publicurl: `${publicurl}/${table}`
|
|||
|
|
};
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async p_set(url, keyvalues, value) { // url = fluence:/fluence/<publickey>/<table>
|
|||
|
|
/*
|
|||
|
|
Set key values
|
|||
|
|
keyvalues: string (key) in which case value should be set there OR
|
|||
|
|
object in which case value is ignored
|
|||
|
|
*/
|
|||
|
|
|
|||
|
|
if (typeof keyvalues === 'string') {
|
|||
|
|
await this.session.request(`HSET ${url} ${keyvalues} ${canonicaljson.stringify(value)}`);
|
|||
|
|
} else {
|
|||
|
|
// Store all key-value pairs without destroying any other key/value pairs previously set
|
|||
|
|
console.assert(!Array.isArray(keyvalues), 'TransportFLUENCE - shouldnt pass an array as the keyvalues');
|
|||
|
|
|
|||
|
|
await Promise.all(
|
|||
|
|
Object.keys(keyvalues).map(hKey => this.session.request(`HSET ${url} ${hKey} ${canonicaljson.stringify(keyvalues[hKey])}`))
|
|||
|
|
);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async p_get(url, keys) {
|
|||
|
|
if (Array.isArray(keys)) {
|
|||
|
|
const result = await this.session.request(`HMGET ${url} ${keys.join(' ')}`);
|
|||
|
|
const data = this.parseRedisResponse(result.asString());
|
|||
|
|
|
|||
|
|
return keys.reduce((store, key, index) => {
|
|||
|
|
const keyValue = data[index];
|
|||
|
|
store[key] = typeof keyValue === "string" ? JSON.parse(keyValue) : keyValue;
|
|||
|
|
|
|||
|
|
return store;
|
|||
|
|
}, {});
|
|||
|
|
} else {
|
|||
|
|
const result = await this.session.request(`HGET ${url} ${keys}`);
|
|||
|
|
let data = this.parseRedisResponse(result.asString());
|
|||
|
|
|
|||
|
|
return typeof data === 'string' ? JSON.parse(data) : data;
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async p_keys(url) {
|
|||
|
|
const result = await this.session.request(`HKEYS ${url}`);
|
|||
|
|
|
|||
|
|
return this.parseRedisResponse(result.asString());
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async p_getall(url) {
|
|||
|
|
const result = await this.session.request(`HGETALL ${url}`);
|
|||
|
|
const dataArray = this.parseRedisResponse(result.asString());
|
|||
|
|
|
|||
|
|
return dataArray.reduce((store, key, index, dataArray) => {
|
|||
|
|
if (index % 2 !== 0) {
|
|||
|
|
return store;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
const keyValue = dataArray[index + 1];
|
|||
|
|
store[key] = typeof keyValue === "string" ? JSON.parse(keyValue) : keyValue;
|
|||
|
|
|
|||
|
|
return store;
|
|||
|
|
}, {});
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
async p_delete(url, keys) {
|
|||
|
|
if (typeof keys === "string") {
|
|||
|
|
await this.session.request(`HDEL ${url} ${keys}`);
|
|||
|
|
} else {
|
|||
|
|
await this.session.request(`HDEL ${url} ${keys.join(' ')}`);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
Transports._transportclasses['FLUENCE'] = TransportFluence;
|
|||
|
|
exports = module.exports = TransportFluence;
|