src/util.js

/**
* utilities, many of which operate against rethinkdb independent of 
* Supermodel constructs.
*/

// @flow

import uuid from 'uuid';
import * as nrser from 'nrser';
const _ = nrser._;
const Promise = nrser.Promise;

// types
import type {
  IRethinkDB,
  IConnection,
} from './types';

/**
* attempt to connect to the database, re-scheduling an attempt with a back-off
* timeout if it fails.
* 
* wraps this all up in a Promise that resolved when the connection succeeds
* or rejects when it runs out of attempts.
*/
function attemptToConnect(
  r: IRethinkDb,
  {
    resolve,
    reject,
    host,
    port,
    db,
    attemptNumber,
    maxAttempts,
    baseTimeout,
    maxTimeout,
  }: {
    resolve: Function,
    reject: Function,
    host: string,
    port: number,
    db: string,
    attemptNumber: number,
    maxAttempts: number,
    baseTimeout: number,
    maxTimeout: number,
  }
): Promise<IConnection> {
  info: `attempt ${ attemptNumber } to connect to db`;
  
  r.connect(
    {host, db, port},
    (error: ?Error, conn: ?IConnection): void => {
      if (error) {
        error: `failed to connect to db on attempt ${ attemptNumber }`,
          {error};
        
        if (maxAttempts > 0 && attemptNumber >= maxAttempts) {
          error: "out of attempts, rejecting."
          
          reject(error);
        } else {
          const timeout = Math.min(timeout * attemptNumber, maxTimeout);
          
          info: `scheduling connection attempt in ${ timeout }ms`;
          
          setTimeout(
            () => {
              attemptToConnect({
                resolve,
                reject,
                attemptNumber: attemptNumber + 1,
                maxAttempts,
                timeout,
              })
            },
            timeout,
          );
        }
      } else {
        info: `db connected on attempt ${ attemptNumber }`;
        resolve(conn);
      }
    }
  );
} // .attemptToConnect

/**
* connect to the db, using `.attemptToConnect` internally.
*
* by default will continue to try connecting every `timeout` milliseconds
* (default 1000ms = 1s).
*/
export function connect(
  r: IRethinkDb,
  {
    host,
    port = 28015,
    db,
    maxAttempts = 0,
    baseTimeout = 1000,
    maxTimeout = 5000,
  }: {
    host: string,
    port?: number,
    db?: string,
    maxAttempts?: number,
    baseTimeout?: number,
    maxTimeout?: number,
  } = {}
): Promise<IConnection> {
  return new Promise((resolve, reject) => {
    attemptToConnect(r, {
      resolve,
      reject,
      host,
      port,
      db,
      attemptNumber: 1,
      maxAttempts,
      baseTimeout,
      maxTimeout,
    });
  });
} // .connect

/**
* makes sure a db exists by attempting to create it and suppressing the
* error if it already exists.
* 
* @return [Promise<boolean>] true if the db was created.
*/
export function ensureDB(
  r: IRethinkDB,
  conn: IConnection,
  dbName: string,
): Promise<boolean> {
  debug: `ensuring db ${ dbName }...`;
  
  return r.dbCreate(dbName).run(conn)
    .then(() => {
      debug: `db ${ dbName } created.`;
      return true;
    })
    .catch((error: Error) => {
      if (error.message.match(/Database\ .*\ already\ exists/)) {
        debug: `db ${ dbName } already present.`;
        return false;
      }
      throw error;
    });
} // .ensureDB

/**
* makes sure a table exists by attempting to create it and suppressing the
* error if it already exists.
* 
* @return [Promise<boolean>] true if the table was created.
*/
export function ensureTable(
  r: RethinkDB,
  conn: IConnection,
  tableName: string,
): Promise<boolean> {
  return r.tableCreate(tableName).run(conn)
    .then(() => true)
    .catch((error: Error) => {
      if (error.message.match(/Table\ .*\ already\ exists/)) {
        return false;
      }
      throw error;
    });
} // .ensureTable

/**
* ensure that an index exists on a table, creating if necessary.
* 
* returns `Promise<boolean>` that resolved `true` if the index was created
* and `false` if it was already there.
*/
export function ensureIndex(
  r: IRethinkDB,
  conn: IConnection,
  tableName: string,
  indexName: string,
  ...indexCreateOptions: Array<*>
): Promise<boolean> {
  debug: `ENSURING index ${ indexName } on table ${ tableName }`, {
    indexCreateOptions
  };
  
  return r
    .table(tableName)
    .indexCreate(indexName, ...indexCreateOptions)
    .run(conn)
    .then((): boolean => {
      debug: `ensuring index ${ indexName } CREATED on table ${ tableName }`;
      return true;
    })
    .catch((error: Error) => {
      if (error.message.match(/Index\ .*\ already\ exists/)) {
        debug: `ensuring index ${ indexName } EXISTED on table ${ tableName }`;
        return false;
      }
      throw error;
    });
} // .ensureIndex