script.js

const path = require('path');
require('dotenv').config({ path: path.dirname(process.argv[1]) + '/.env' });
const snowflake = require('snowflake-sdk');
const axios = require('axios');
const qs = require('qs');
const datetime = require('node-datetime');
const { program } = require('commander');
const Datastore = require('@seald-io/nedb');

const logsDb = new Datastore({ filename: path.dirname(process.argv[1]) + '/logs/logs.db', autoload: true, timestampData: true });
logsDb.ensureIndex({ fieldName: "createdAt" });
const yesterday = yesterdayDateString();
const trafsysUrl = 'https://portal.trafnet.com/rest/';

/**
 * A log object that collects info related to a run of this program.
 * @typedef {Object} RunInfo
 * @property {string} AccessToken - The access token used to get Trafsys data. Will be reused until it expires.
 * @property {Date} AccessTokenExpiresAt - The time at which the access token expires.
 * @property {string} FromDate - The From date used for this run, in YYYY-MM-DD format.
 * @property {string} ToDate - The To date used for this run, in YYYY-MM-DD format.
 * @property {number?} Records - The number of records written to the Snowflake db.
 */

/**
 * Generates a RunInfo object for the current run.
 * @returns {Promise<RunInfo>}
 */
async function getRunInfo() {
  // helper function to call nedb cursor methods using async/await syntax
  let execAsync = cursor => new Promise(
    (resolve, reject) => cursor.exec((err, result) => err ? reject(err) : resolve(result))
  );
  // use sort and limit to get most recently saved log object
  let previousRun = await execAsync(logsDb.findOne({}).sort({ createdAt: -1 }).limit(1));
  let currentRun = {};
  if (previousRun) {
    let expiresAt = datetime.create(previousRun.AccessTokenExpiresAt);
    let nowish = datetime.create();
    // Offset by 5 minutes to give some wiggle room (technical term)
    nowish.offsetInHours(-1/12);
    // .getTime() converts the object to a timestamp for comparison
    if (expiresAt.getTime() > nowish.getTime()) {
      currentRun.AccessToken = previousRun.AccessToken;
      currentRun.AccessTokenExpiresAt = previousRun.AccessTokenExpiresAt;
    }
  }
  if (!currentRun.AccessToken) {
    let tokenData = await getAccessToken();
    currentRun.AccessToken = tokenData.access_token;
    currentRun.AccessTokenExpiresAt = new Date(tokenData[".expires"]);
  }
  program
    .option('-f, --from <date>', 'From Date (YYYY-MM-DD)', previousRun?.ToDate || yesterday)
    .option('-t, --to <date>', 'To Date (YYYY-MM-DD)', yesterday);
  program.parse();
  let opts = program.opts();
  currentRun.FromDate = opts.from;
  currentRun.ToDate = opts.to;
  return currentRun;
}

/**
 * Gets a fresh access token from TrafSys.
 *
 * @returns {Promise<{access_token: string, ".expires": string}>}
 */
async function getAccessToken() {
  let tokenResponse = await axios.post(trafsysUrl + 'token', qs.stringify({
    username: process.env.TRAFSYS_USER,
    password: process.env.TRAFSYS_PASSWORD,
    grant_type: 'password'
  }), {
    headers: {
      'Content-Type': 'application/x-www-form-urlencoded'
    }
  });
  return tokenResponse.data;
}

/**
 * Checks if all the required environment variables are present.
 * If they are not, display an error and quit.
 */
function checkEnv() {
  let keys = [
    'SNOWFLAKE_ACCOUNT',
    'SNOWFLAKE_USER',
    'SNOWFLAKE_PASSWORD',
    'SNOWFLAKE_WAREHOUSE',
    'SNOWFLAKE_DATABASE',
    'SNOWFLAKE_SCHEMA',
    'TRAFSYS_USER',
    'TRAFSYS_PASSWORD'
  ];
  let missingKeys = keys.filter(key => !(key in process.env));
  if (missingKeys.length === 0) return;
  console.error('Missing required environment variables: ' + missingKeys.join(', '));
  process.exit();
}

/**
 * Creates a Snowflake connection and resolves once it's connected.
 * @returns {Promise<snowflake.Connection>}
 */
function createConnection() {
  return new Promise((resolve, reject) => {
    snowflake.configure({
      logLevel: 'ERROR',
      logFilePath: path.dirname(process.argv[1]) + '/logs/snowflake.log',
    });
    let connection = snowflake.createConnection({
      account: process.env.SNOWFLAKE_ACCOUNT,
      username: process.env.SNOWFLAKE_USER,
      password: process.env.SNOWFLAKE_PASSWORD,
      warehouse: process.env.SNOWFLAKE_WAREHOUSE,
      database: process.env.SNOWFLAKE_DATABASE,
      schema: process.env.SNOWFLAKE_SCHEMA,
      role: process.env.SNOWFLAKE_ROLE // optional
    });
    connection.connect((err, conn) => {
      if (err) reject(err);
      else resolve(conn);
    });
  });
}

/**
 * Destroys (closes) a Snowflake connection.
 * @param {snowflake.Connection} connection
 * @returns {Promise<void>}
 */
function closeConnection(connection) {
  return new Promise((resolve, reject) => {
    connection.destroy((err) => err ? reject(err) : resolve());
  });
}

/**
 * Promise wrapper around connection.execute.
 * @param {snowflake.Connection} connection
 * @param {string} sqlText
 * @param {Array} [binds] - Either a flat array of bind values, or an array of
 *   arrays for a bulk/array-bind operation (e.g. multi-row inserts).
 * @returns {Promise<any[]>} The rows returned by the statement, if any.
 */
function executeAsync(connection, sqlText, binds) {
  return new Promise((resolve, reject) => {
    connection.execute({
      sqlText,
      binds,
      complete: (err, stmt, rows) => {
        if (err) {
          // The driver's error objects are often light on context (no
          // indication of which statement failed, or which bind values
          // were involved). Attach that here so logError has something
          // useful to print, without altering err's own fields.
          err.sqlText = sqlText;
          err.binds = binds;
          err.queryId = stmt?.getStatementId?.();
          reject(err);
        } else {
          resolve(rows);
        }
      }
    });
  });
}

/**
 * Logs an error with as much diagnostic detail as is available: the
 * stack trace, any Snowflake-specific fields (SQL state, error code,
 * query ID), and, if executeAsync attached them, the offending SQL text
 * and bind values.
 * @param {string} context - A short label identifying where this occurred.
 * @param {any} e - The caught error.
 */
function logError(context, e) {
  console.error(`--- Error in ${context} ---`);
  console.error(e?.stack || e?.message || e);
  let extras = {};
  for (let key of ['code', 'sqlState', 'queryId', 'response']) {
    if (e?.[key] !== undefined) extras[key] = e[key];
  }
  if (Object.keys(extras).length > 0) {
    console.error('Details:', extras);
  }
  if (e?.sqlText) {
    console.error('SQL:', e.sqlText);
  }
  if (e?.binds) {
    // Binds can be a large array of arrays (one per row); printing all of
    // them would flood the log, so just show the first few rows plus a
    // count. That's usually enough to spot something like an undefined
    // value or a wrong column count.
    let binds = e.binds;
    let sample = Array.isArray(binds) && Array.isArray(binds[0])
      ? binds.slice(0, 3)
      : binds;
    console.error(`Binds (showing ${sample.length} of ${Array.isArray(binds) ? binds.length : 1}):`, sample);
  }
  console.error('---');
}

/**
 * Creates the ULS_TRAFSYS_DATA table if it does not exist.
 * @param {snowflake.Connection} connection - The database connection.
 */
async function ensureTableExists(connection) {
  await executeAsync(connection,
    `create table if not exists ULS_TRAFSYS_DATA
     (
       SiteCode varchar(100),
       Location varchar(100),
       IsInternal number(1),
       PeriodEnding timestamp_ntz,
       Ins number,
       Outs number,
       primary key(SiteCode, Location, PeriodEnding)
     )`
  );
}

/**
 * A TrafSys data record.
 * @typedef {Object} DataRecord
 * @property {string} SiteCode - The alphanumeric code that identifies the site within the organization.
 * @property {string} Location - The name of the location where the sensors are counting.
 * @property {number} IsInternal - Indicates (using 0 or 1) whether this is an internal location.
 * @property {string} PeriodEnding - The end of the hour-long time period this record corresponds to.
 * @property {number} Ins - The in counts for that time period and location.
 * @property {number} Outs - The out counts for that time period and location.
 */

/**
 * Retrieves TrafSys data from the REST API.
 * @param {RunInfo} runInfo - The run information for the current run.
 * @returns {Promise<DataRecord[]>} Data pulled from the api and given a RecordId.
 */
async function getTrafsysData(runInfo) {
  let dataResponse = await axios.get(trafsysUrl + 'api/traffic', {
    params: {
      SiteCode: '',
      IncludeInternalLocations: true,
      DataSummedByDay: false,
      DateFrom: runInfo.FromDate,
      DateTo: runInfo.ToDate,
    },
    headers: {
      'Authorization': 'Bearer ' + runInfo.AccessToken
    }
  });
  let data = dataResponse.data;
  // check that the response actually contains data
  if (!data?.[Symbol.iterator])
  {
    throw new Error("Bad response from Trafsys: " + toString(dataResponse));
  }
  for (let record of data) {
    // Cast to a number for consistency with the Snowflake NUMBER(1) column
    record.IsInternal = +record.IsInternal;
  }
  runInfo.Records = data.length;
  return data;
}

/**
 * @returns {string} Yesterday's date formatted as a YYYY-MM-DD string.
 */
function yesterdayDateString() {
  var date = datetime.create();
  date.offsetInDays(-1);
  return date.format('Y-m-d');
}

/**
 * Inserts the TrafSys data into the database, upserting on
 * (SiteCode, Location, PeriodEnding).
 *
 * Snowflake has no equivalent of Oracle's "insert ... exception when
 * dup_val_on_index then update" pattern, and MERGE doesn't support
 * array/bulk binding directly. Instead, this bulk-loads the incoming
 * rows into a temporary staging table (using array binds, which IS
 * supported for plain INSERTs) and then MERGEs from staging into the
 * target table in a single statement.
 *
 * Note: array/bulk binding only supports plain `?` placeholders mapped
 * directly to columns — a bind variable can't be wrapped in a function
 * call (e.g. TO_TIMESTAMP_NTZ(?)) the way it can with a normal single-row
 * bind. PeriodEnding is bound as its raw ISO string and relies on
 * Snowflake's default TIMESTAMP_INPUT_FORMAT (AUTO) to cast it into the
 * TIMESTAMP_NTZ column.
 *
 * @param {snowflake.Connection} connection - The database connection.
 * @param {DataRecord[]} data
 */
async function insertData(connection, data) {
  if (data.length === 0) return;

  // Fresh staging table for this run. TEMPORARY tables are session-scoped,
  // so this is automatically cleaned up when the connection closes.
  await executeAsync(connection,
    `create or replace temporary table ULS_TRAFSYS_DATA_STAGING
     (
       SiteCode varchar(100),
       Location varchar(100),
       IsInternal number(1),
       PeriodEnding timestamp_ntz,
       Ins number,
       Outs number
     )`
  );

  // Array binds: one array of values per row, in column order.
  // `?? null` matters here: the Snowflake driver can bind `null` but not
  // `undefined` (it raises "Bind variable ? not set"), and TrafSys
  // occasionally omits a field on a given record.
  let binds = data.map(record => [
    record.SiteCode ?? null,
    record.Location ?? null,
    record.IsInternal ?? null,
    record.PeriodEnding ?? null,
    record.Ins ?? null,
    record.Outs ?? null
  ]);

  await executeAsync(connection,
    `insert into ULS_TRAFSYS_DATA_STAGING
     (SiteCode, Location, IsInternal, PeriodEnding, Ins, Outs)
     values (?, ?, ?, ?, ?, ?)`,
    binds
  );

  await executeAsync(connection,
    `merge into ULS_TRAFSYS_DATA t
     using ULS_TRAFSYS_DATA_STAGING s
       on t.SiteCode = s.SiteCode
      and t.Location = s.Location
      and t.PeriodEnding = s.PeriodEnding
     when matched then update set
       t.Ins = s.Ins,
       t.Outs = s.Outs
     when not matched then insert
       (SiteCode, Location, IsInternal, PeriodEnding, Ins, Outs)
       values (s.SiteCode, s.Location, s.IsInternal, s.PeriodEnding, s.Ins, s.Outs)`
  );
}

/**
 * Returns a promise that resolves after one second.
 * @returns {Promise<void>}
 */
function waitASecond() {
  return new Promise(resolve => setTimeout(resolve, 1000));
}

/**
 * Run the program, pulling data from TrafSys and inserting it into the database.
 */
async function run() {
  checkEnv();
  let connection;
  try {
    connection = await createConnection();
    await ensureTableExists(connection);
    let runInfo = await getRunInfo();
    let trafsysData;
    try {
      trafsysData = await getTrafsysData(runInfo);
    }
    catch (e) {
      if (e.isAxiosError && e.response?.status == 401) {
        // wait a second to prevent "429 Too Many Requests"
        await waitASecond();
        let tokenData = await getAccessToken();
        runInfo.AccessToken = tokenData.access_token;
        runInfo.AccessTokenExpiresAt = new Date(tokenData[".expires"]);
        trafsysData = await getTrafsysData(runInfo);
      } else {
        throw e;
      }
    }
    await insertData(connection, trafsysData);
    logsDb.insert(runInfo);
  }
  catch (e) {
    logError('run', e);
  }
  finally {
    if (connection) {
      await closeConnection(connection);
    }
  }
}

run();