Distraught Web Server

I was distraught!!

Distraught is a wrapper around a Node.js Express server that exposes an HTTPServer for web requests, a CronServer for functions that need to be called at set intervals, and a WorkerServer to handle long requests.

Assumptions

• You have a webserver running Node.js
• Postgresql as the datastore
• Redis as the caching layer and sockets
• RabbitMQ for worker jobs

Libraries

• ExpressJS
• Mariner - Vanilla SQL Database Migrations
• Heretic - Queueing / Dequeueing Jobs
• Google Cloud Storage - File Storage
• Sentry - Error Handling
• Cron - For Timed Jobs
• Axios - For API Requests
• Sendgrid - Sending Emails
• Twilio - Text Messaging
• Socket.IO - Websockets
• Swagger - API Documentation

Expected Migrations / Tables In Order To Use WorkerServer

This does require some migrations to be ran, however this server does -not- run the migrations on startup. If you are using Distraught for the first time, please run the following migration:

CREATE TABLE heretic_jobs (
  id SERIAL NOT NULL PRIMARY KEY,
  queue_name text NOT NULL,
  status text DEFAULT 'pending',
  payload jsonb,
  attempt_logs jsonb[] DEFAULT '{}',
  max_attempts int NOT NULL DEFAULT 1,
  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now(),
  last_attempted_at timestamptz
);

CREATE INDEX ON heretic_jobs (queue_name);
CREATE INDEX ON heretic_jobs (status);

CREATE FUNCTION heretic_updated_at_timestamp() RETURNS TRIGGER AS $$
BEGIN
  NEW.updated_at = now();
  RETURN NEW;
END;
$$ LANGUAGE 'plpgsql';

CREATE TRIGGER update_heretic_jobs_updated_at
  BEFORE UPDATE ON heretic_jobs
  FOR EACH ROW EXECUTE PROCEDURE heretic_updated_at_timestamp();

Processes

The framework is setup to run three processes: web, crons, and workers.

• Web
  • Will boot up the Express server
• Crons
  • These are processes that run in the background at set intervals
• Workers
  • Workers are background processes that may take quite a bit of time to complete, so they are enqueued to a RabbitMQ server via Heretic, a simple RabbitMQ interface

Logging

• Supports sending all server logs to Logentries if a LOGENTRIES_TOKEN is present.
• Supports sending uncaught/unhandled errors to Sentry if a SENTRY_DSN is present.

Logging Errors

const {logErr} = require('distraught');

logErr(new Error('Whoa..That\'s not good'), {
  var1: val1,
  var2: val2,
});

Database

• Utilizes Knex to handle your queries

Setting Up Your Connections

const d = require('distraught');

d.init({
  db: {
    r: { connection: process.env.READONLY_DATABASE_URL },
    rw: { connection: process.env.READWRITE_DATABASE_URL },
  },
  heretic: {
    default: { db: "r", connection: process.env.AMQP_URL },
  },
  cache: {
    default: { connection: process.env.REDIS_URL },
  },
  captureUncaught: true,
  captureUnhandled: true,
});

Querying The Database

const {db, toCamelCase, createOne} = require('distraught');

function fetchUsers() {
  return db.r('users')
    .column(['*'])
    .limit(1000)
    .then(toCamelCase);
}

function addUser(user) {
  return createOne(db.rw, 'users', user)
    .then(toCamelCase);
}

HTTPServer

const {httpServer, init} = require('distraught');

init({ 
  cache: {
    default: { connection: process.env.REDIS_URL },
  },
});

const homeController = require('./controllers/home');

const server = httpServer({
  publicPath: path.join(__dirname, 'public'),
  viewPath: path.join(__dirname, 'views'),
  findUserById(id: number) {
    return cache.default.getOrSet(`user-${id}`, fetchUserById.bind(null, id)); // Needed for passport middleware
  },
});

server.app.use((req, res, next) => {
  // ...some middleware/plugin logic
  next();
});

/* WEB ROUTES */
server.app.get('/', homeController.get);

authController.setAuthRoutes(server.app, server.passport);

server.start();

Templates

By default, HTTPServer will render pug templates, but you can change the view engine to whatever you want during instantiation.

Adding/Changing View Engines

const {httpServer} = require('distraught');

const server = httpServer({
  viewEngine: 'jsx',
});

server.app.engine('jsx', (filePath, options, callback) => { 
  const html = templateRenderFn(filePath, options);
  callback(null, html);
});

Swagger

To enable Swagger:

const server = httpServer({
  swaggerConfig: {
    appRoot: __dirname,
    yamlPath: path.join(__dirname, 'api/swagger/swagger.yaml'),
  },
});

Example of Creating API/Config Folders and Using the Swagger Editor Swagger - Getting Started

WorkerServer

// Make sure the Heretic database migration has run

const {init, workerServer, MINUTE, heretic, chalk, log} = require('distraught');

init({
  db: {
    r: { connection: process.env.READONLY_DATABASE_URL },
    rw: { connection: process.env.READWRITE_DATABASE_URL },
  },
  heretic: {
    default: { db: "r", connection: process.env.AMQP_URL },
  },
});


function testDequeue(job, message, done) {
  log(chalk.yellow('Dequeueing job: Test queue'));
  return Promise.resolve()
    .then(done);
}

function queueJob() {
  heretic.default.enqueue('test.dequeue', {});
  setTimeout(() => {
    queueJob();
  }, 5000);
}

function startWorkerServer() {
  const debug = process.env.WORKER_DEBUG;
  const workers = workerServer({
    heretic: heretic.default,
    requiredEnv: [],
    queues: [
      {name: 'test.dequeue', concurrency: 3, handler: testDequeue, isEnabled: process.env.NODE_ENV === 'development', alertAt: MINUTE, killAt: MINUTE * 2, debug},
    ],
  });
  workers.start();
};

queueJob();
startWorkerServer();

Enqueueing jobs

import {heretic} from 'distraught';

heretic.default.enqueue('test.dequeue', {});

Crons

const {cronServer, log, chalk} = require('distraught');

exports.startCronServer = () => {
  cronServer({
    crons: [
      {
        name: 'Ping',
        cronTime: '* * * * * *', // Every second
        onTick() {
          log(chalk.green('Pong'));
        },
      },
    ],
  });
};

Caching

Caching Individual Functions

Getting value from cache by key, or setting it via a function

  const {init, cache, MINUTE} = require('distraught');

  init({
    cache: {
      default: { connection: process.env.REDIS_URL },
    },
  });

  const getValueFn = () => {
    return someFuncReturningData();
  }; // Can be a scalar, function returning a scalar, or function returning a Promise
  const ttl = MINUTE * 3;

  function getUsers() {
    return cache.default.getOrSet('all-users', getValueFn, ttl)
      .then((users) => console.log(users));
  }

  await getUsers(); // Cache missed
  await getUsers(); // Cache hit

Invalidating Keys

The below example will remove all-users from the cache

  const {cache} = require('distraught');
  cache.default.invalidate('all-users');

Thanks

Thanks to Hackathon Starter for a lot of inspiration