src/RethinkAdapter.js
// deps
import * as nrser from 'nrser';
const Promise = nrser.Promise;
const _ = nrser._;
import { Record, Adapter } from '@nrser/supermodel';
// package
import * as util from './util';
// types
import t from '@nrser/supermodel/lib/types';
import type {
Integer,
PositiveInteger,
} from 'nrser/lib/types/number';
import type {
Id,
Document
} from '@nrser/supermodel/lib/types';
import type {
IRethinkDb,
IConnection,
ITable,
} from './types';
/**
* Supermodel Adapter for RethinkDB.
*/
export class RethinkAdapter extends Adapter {
/**
* @param {rethinkdb} [rethinkdb]
* the module returned from require('rethinkdb').
*
* we need to pass it in so that we don't end up using a different
* instantiation than the one that created the connection.
*
* @param {rethinkdb.Net.Connection} [conn]
* connection to the rethink database.
*/
constructor(rethinkdb, conn) {
super();
this.r = rethinkdb;
this.conn = conn;
} // constructor
// rethink-specific utils
// ========================================================================
//
/**
* get the table name for a Record class; uses the tcomb type name.
*/
tableName(recordClass: Class<Record>): string {
return recordClass.meta.name;
}
/**
* get the rethink table for a Record class.
*/
table(recordClass: Class<Record>): RethinkTable {
return this.r.table(this.tableName(recordClass));
}
/**
* handle write (insert, replace) response from RethinkDB, which has an
* `errors` field with the number of write errors.
*
* convert it into a Promise rejection if present.
*/
handleWriteResponse(response): Promise<void> {
if (response.errors > 0) {
return Promise.reject(new Error(response.first_error));
}
return Promise.resolve();
} // .handleWriteResponse
/**
* ensure the DB is setup for the record classes.
*
* includes:
*
* 1. ensure the database exists.
* 2. for each Record class:
* 1. ensure the table exists.
* 2. ensure the indexes exist.
*/
ensureDBSetup(recordClasses: Array<Class<Record>>): Promise<void> {
return (
util.ensureDB(this.r, this.conn, this.conn.db)
.then(() => this.ensureTables(recordClasses))
.then(() => this.ensureIndexes(recordClasses))
.then(() => undefined) // return undefined
); // return
} // #ensureDBSetup
/**
* ensure tables for the Record classes have been created.
*
* returns a Promise of an object with table names mapped to whether the
* table had to be created or not.
*/
ensureTables(
recordClasses: Array<Class<Record>>,
): Promise<{[tableName: string]: boolean}> {
return (
Promise.props(
_.chain(recordClasses)
.map((recordClass) => {
const tableName = this.tableName(recordClass);
debug: `ensuring table ${ tableName }...`;
return [
tableName,
util.ensureTable(this.r, this.conn, tableName),
];
})
.fromPairs()
.value()
) // Promise.props
); // return
} // ensureTable
/**
* ensure all the indexes declared in the Record class exist, creating if
* necessary. if `wait` is true, uses `rethinkdb.indexWait` to wait for the
* index to be ready.
*
* returns a `Promise<void>` that resolves when all done, or rejects on error.
*/
ensureIndexes(
recordClasses: Class<Record> | Array<Class<Record>>,
{
wait = true,
} : {
wait: boolean,
} = {}
): Promise<
{[recordName: string]: {[indexName: string]: boolean}} |
{[indexName: string]: boolean}
> {
debug: "ensuring indexes...";
// if we were passed and array of Record classes, return a promise that
// maps their meta names to the results of ensuring the indexes on
// each one.
if (Array.isArray(recordClasses)) {
return (
Promise.props(
_.chain(recordClasses)
.map((recordClass) => {
return [
recordClass.meta.name,
this.ensureIndexes(recordClass, {wait})
];
})
.fromPairs()
.value() // end chain
) // Promise.props
); // return
} // if isArray
// now it should be a single record class
const recordClass: Class<Record> = recordClasses;
const tableName: string = this.tableName(recordClass);
debug: `ensuring indexes for ${ recordClass.meta.name }`;
// return a promise mapping index names to whether they were
// created or not
return (
Promise.props(
// take the index meta data
_.chain(recordClass.meta.indexes)
// pick the ones with truth-y values
.pickBy(
(shouldIndex: boolean, indexName: string) => !!shouldIndex
)
// get the keys, which are the index names
.keys()
// and map each to the ensureIndex promise
.map((indexName: string): [string, Promise<boolean>] => {
let promise = util.ensureIndex(
this.r,
this.conn,
tableName,
indexName
);
return [indexName, promise];
})
.fromPairs()
.value()
) // Promise.props
.tap((created: {[indexName: string]: boolean}) => {
debug: "indexes ENSURED", created;
if (wait) {
debug: `waiting for indexes on ${ tableName }...`;
return this.table(recordClass).indexWait().run(this.conn)
.then(() => {
debug: `indexes ready on ${ tableName }.`;
});
}
})
); // return
} // #ensureIndexes
// Adapter API implementations
// =========================================================================
//
/**
* @private
* @override
*/
_get(
recordClass: Class<Record>,
id: Id,
options: Object,
): Promise<Document> {
let query = this.table(recordClass).get(id);
if (options.join) {
query = query.merge(function(instance) {
return _.mapValues(options.join, ({recordClass, idProp}) => {
return this.table(recordClass).get(instance(idProp));
})
})
}
return query.run(this.conn);
}
/**
* @private
* @override
*/
_need(
recordClass: Class<Record>,
id: Id,
options: Object,
): Promise<Document> {
return this.get(recordClass, id, options)
.then((doc: ?Document) => {
if (doc === null) {
throw new nrser.NotFoundError(
`no ${ recordClass.name } found with id ${ id }`
);
}
return doc;
});
} // #_need
/**
* @private
* @override
*
* @param {Object} options
* @param {boolean} options.compact
* if true, the results should have keys with null values omitted.
*/
_getByIds(
recordClass: Class<Record>,
ids: Array<Id>,
options: Object,
): Promise<{[id: Id]: ?Record}> {
const compacted = this
.table(recordClass)
.getAll(...ids)
.run(this.conn)
.then((cursor) => cursor.toArray())
.then((docs) => {
return _.keyBy(docs, 'id');
});
if (options.compact) {
return compacted;
}
return compacted.then((byIds) => {
_.each(ids, (id) => {
if (!_.has(byIds, id)) {
byIds[id] = null;
}
});
return byIds;
});
} // #_getByIds
/**
* @private
* @override
*/
_needByIds(
recordClass: Class<Record>,
ids: Array<Id>,
options: Object = {},
): Promise<{[id: Id]: Record}> {
return this
// call _getByIds with compact=true so that it doesn't add null entries
._getByIds(recordClass, ids, {compact: true})
// then check if we got all of them
.then((byIds) => {
if (_.size(byIds) !== ids.length) {
// something's missing, throw up
const missingIds = _.difference(ids, _.keys(byIds));
throw nrser.NotFoundError.squish(`
could not find ${ recordClass.meta.name } with ids
${ missingIds }
`);
}
return byIds;
});
} // #_needByIds
/**
* @private
* @override
*/
_all(
recordClass: Class<Record>,
options: Object
): Promise<{[id: Id]: Document}> {
return this
.table(recordClass)
.run(this.conn)
.then((cursor) => {
return cursor.toArray();
})
.then((docs) => {
return _.keyBy(docs, 'id');
});
}
/**
* internal implementation of {#slice} called after arguments have been
* validated and normalized.
*/
_slice(
recordClass: Class<Record>,
{
start,
end,
}: {
start: Integer,
end?: Integer,
}
): Promise<Array<Document>> {
// supermodel.Adapter#slice takes care of argument validation and
// short-circuiting empty slices. when we get here we know:
//
// - end and limit are not both defined
// - if end is defined
// - it is an integer greater than start
// - end - start > 0
//
if (start < 0) {
// we're starting the slice at the end of the array, which rethink
// doesn't natively support, so we need to convert it into a slice from
// the beginning on a reverse-ordered cursor, then reverse the results.
// the reserve end is just the negation of the (negative) start (making
// it positive)
const reverseEnd = -1 * start;
// the reverse end is either 0 if no end was provided, or the negation
const reserveStart = t.Nil.is(end) ? 0 : -1 * end;
return this.table(recordClass)
.orderBy({index: this.r.desc('createdAt')})
.slice(reserveStart, reverseEnd)
.run(this.conn)
.then(cursor => cursor.toArray())
.then(docs => docs.reverse());
}
return this.table(recordClass)
.orderBy({index: this.r.asc('createdAt')})
.slice(start, end)
.run(this.conn)
.then((cursor) => {
return cursor.toArray();
});
} // #_slice
/**
* @private
* @override
*
* @param {Object} options
* for future use.
*/
_sample(
recordClass: Class<Record>,
number: PositiveInteger,
options: Object,
): Promise<Array<Document>> {
return this
.table(recordClass)
.sample(number)
.run(this.conn);
} // #_sample
/**
* insert a Record into the database.
*
* returns a new Record instance that reflects the changes (setting
* `updatedAt`).
*/
insert(
record: Record,
): Promise<Record> {
const data = record.makeObjectForWrite();
return this.r
.table(this.tableName(record.constructor))
.insert(data)
.run(this.conn)
.then(this.handleWriteResponse.bind(this))
.then(() => new record.constructor(data));
} // #insert
/**
* replace a Record in the database.
*
* returns a new Record instance that reflects the changes (setting
* `updatedAt`).
*/
replace(
record: Record,
): Promise<Record> {
const data = record.makeObjectForWrite();
return this.r
.table(this.tableName(record.constructor))
.get(record.id)
.replace(data)
.run(this.conn)
.then(this.handleWriteResponse.bind(this))
.then(() => new record.constructor(data));
} // #replace
/**
* internal driver-specific implementation of {#update} called after
* arguments have been validated and normalized.
*
* the updated Record instance is kept in {#update} and resolved from there;
* this method only need to resolve `void` on success or reject with an
* Error on failure.
*/
_update(
record: Record,
props: Object,
{
check = true,
} : {
check?: boolean,
} = {},
): Promise<void> {
let query = this.r
.table(this.tableName(record.constructor))
.get(record.id);
if (check) {
// we need to check that the rev hasn't changed from the version
// we are updating
query = query.update((data) => {
return this.r.branch(
data('rev').eq(record.rev),
props,
this.r.error("rev in database does not match record rev.")
);
});
} else {
// we don't care what rev is in the db, just blast it
query = query.update(props);
}
return query.run(this.conn)
.then(this.handleWriteResponse.bind(this));
} // #_update
// adaptations
// ===========
} // RethinkAdapter
