@corva/node-sdk v8.5.0

node-sdk

Corva node-sdk is a framework for building Corva DevCenter apps.

Contents

• Requirements
• Installation
• App types
• Event
• Api
• Cache
• Contributing

Requirements

Node.js 12+

Installation

$ npm i @corva/node-sdk

App types

There are 4 app types that you can build:

1. stream - works with real-time data
2. scheduled - works with data at defined schedules/intervals (e. g. once an hour)
3. task - works with data on-demand

Note: Use type hints like those in the examples below for better support from editors and tools.

Stream

/* eslint-disable camelcase */
import { StreamEventHandler, Corva } from '@corva/node-sdk';

const streamApp: StreamEventHandler<any, void> = async (event, { api }) => {
    // get some data from api
    const { body: drillstrings } = await api.request<DrillStringRecord[]>(
        'api/v1/data/corva/data.drillstring/',
        {
            searchParams: {
                query: JSON.stringify({ asset_id: event.asset_id }),
                sort: JSON.stringify({ timestamp: 1 }),
                limit: 1,
            },
        }
    );

    // do some calculations/modifications to the data
    const total = drillstrings.reduce((acc, drillstring) => acc + drillstring.data.bit_depth, 0);
    const average = total / drillstrings.length;

    // save the data to private collection
    await api.request('api/v1/data/my_provider/my_collection/', {
        method: 'POST',
        json: [
            {
                timestamp: Date.now() / 1000,
                company_id: 42,
                asset_id: 42,
                version: 1,
                data: { average },
            },
        ],
    });
};

exports.handler = new Corva().stream(streamApp, {
  appKey: 'app-key',
  name: 'app-name',
  provider: 'app-provider',
});

Corva#stream provides an optional parameter:

• filteringMode - set to timestamp or depth to clear event data with previously processed timestamp or measured_depth to prevent duplicate records.

Scheduled

import { ScheduledEventHandler, Corva } from '@corva/node-sdk';

const scheduledApp: ScheduledEventHandler<{ status: string }> = async (
    event,
    { api, logger }
) => {
  const queryTo = Math.floor(event.schedule_start / 1000);
  const { body: records } = await api.request('api/v1/data/my_provider/my_collection/', {
      method: 'GET',
      searchParams: {
          query: JSON.stringify({
              asset_id: event.asset_id,
              timestamp: {
                  $gt: queryTo - event.interval,
                  $lte: queryTo,
              },
          }),
          sort: JSON.stringify({ timestamp: 1 }),
          limit: 100,
          skip: 0,
      },
  });

  logger.info(records);

  return { status: 'OK' };
};

export const handler = new Corva().scheduled(scheduledApp);

Task

/* eslint-disable camelcase */
import { Corva, TaskEventHandler } from '@corva/node-sdk';

interface TaskProps {
    some: { input: string };
}

interface DatasetRecord {
    data: { foo: string };
}

interface TaskOutput {
    opts: string;
    some: {
        output: DatasetRecord[];
    };
}

const timezonesShift = {
    'America/Chicago': 6,
};

const taskApp: TaskEventHandler<TaskProps, TaskOutput> = async (event, { api }) => {
    // get some data from api
    const settings = await api.getAppSettings();
    // now - 7 days + timezone shift
    const timestampToLoadDataFrom =
        Date.now() / 1000 - 7 * 24 * 60 * 60 + timezonesShift[settings.timezone] || 0;

    const results = await api.getDataset<DatasetRecord>({
        dataset: 'some-dataset',
        query: {
            timestamp: { $gte: timestampToLoadDataFrom },
        },
        sort: { timestamp: 1 },
        limit: 1000,
        fields: ['data.foo'],
    });

    return {
        opts: event.properties.some.input,
        some: { output: results },
    };
};

export const handler = new Corva().task(taskApp);

Event

An event is an object that contains data for an app function to process. event instance is inserted automatically as the first parameter to each app type. There are different event types for every app type: StreamLambdaEvent, ScheduledLambdaEvent, and Task.

Api Client

Apps might need to communicate with the Corva Platform API and Corva Data API. This SDK provides an CorvaDataSource class, which is node-api-client and based on got

Examples:

import { ScheduledEventHandler, Corva } from '@corva/node-sdk';

const scheduledApp: ScheduledEventHandler<{ status: string }> = async (event, { api }) => {
  // Corva API calls
  await api.request('/v2/pads', {
    searchParams: { param: 'val' },
  });
  await api.request('/v2/pads', {
    method: 'POST',
    json: { key: 'val' },
  });
  await api.request('/v2/pads', {
    method: 'PUT',
    json: { key: 'val' },
  });
  await api.request('/v2/pads', {
    method: 'DELETE',
  });

  // Corva Data API calls
  await api.request('api/v1/data/provider/dataset/', {
    searchParams: { param: 'val' },
  });
  await api.request('api/v1/data/provider/dataset/', {
    method: 'POST',
    json: { key: 'val' },
  });
  await api.request('api/v1/data/provider/dataset/', {
    method: 'PUT',
    json: { key: 'val' },
  });
  await api.request('api/v1/data/provider/dataset/', {
    method: 'DELETE',
  });

  return { status: 'OK' };
};

export const handler = new Corva().scheduled(scheduledApp);

Cache

Apps might need to share some data between runs. The SDK provides a Cache class that allows you to store, load, and do other operations with data. Cache instance is inserted automatically to stream and scheduled apps. Note: task apps don't get a Cache parameter as they aren't meant to store data between invokes.

Cache uses a dict-like database, so the data is stored as key:value pairs. key should be of string type, and value can have any of the following types: string, Buffer, number,any[].

Examples:

See source codes.

Store and load:

import { ScheduledEventHandler, Corva } from '@corva/node-sdk';

const scheduledApp: ScheduledEventHandler<{ status: string }> = async (event, { cache }) => {
  await cache.store({ key: 'val' });

  // val
  const res = await cache.load('key');

  return { status: 'OK' };
};

export const handler = new Corva().scheduled(scheduledApp);

Store and load multiple:

import { ScheduledEventHandler, Corva } from '../../lib';

const scheduledApp: ScheduledEventHandler<{ status: string }> = async (event, { cache }) => {
  await cache.store({ key1: 'val1', key2: 'val2', key3: 'val3' });

  // { key1: 'val1', key2: 'val2' }
  const res = await cache.loadAll();

  return { status: 'OK' };
};

export const handler = new Corva().scheduled(scheduledApp);

Delete and delete all:

import { ScheduledEventHandler, Corva } from '@corva/node-sdk';

const scheduledApp: ScheduledEventHandler<{ status: string }> = async (event, { cache }) => {
  await cache.store({ key1: 'val1', key2: 'val2', key3: 'val3' });
  // cache: { key1: 'val1', key2: 'val2', key3: 'val3' }

  await cache.delete('key1');
  // cache: { key2: 'val2', key3: 'val3' }

  await cache.deleteAll();
  // cache: {}

  return { status: 'OK' };
};

export const handler = new Corva().scheduled(scheduledApp);

Expiry, TTL, and exists.

Note: by default, Cache sets an expiry to 60 days.

import { ScheduledEventHandler, Corva } from '../../lib';

const sleep = (seconds: number) => new Promise((resolve) => setTimeout(resolve, seconds * 1000));

const scheduledApp: ScheduledEventHandler<{ status: string }> = async (event, { cache }) => {
  await cache.store({ key1: 'val1', key2: 'val2', key3: 'val3' }, 60 /* seconds */);
  // cache: { key1: 'val1', key2: 'val2', key3: 'val3' }

  await cache.ttl(); // 60 seconds
  await cache.pttl(); // 60000 milliseconds
  await cache.exists(); // true

  await sleep(60);
  // cache: {}

  await cache.exists(); // false
  await cache.ttl(); // -2: doesn't exist
  await cache.pttl(); // -2: doesn't exist

  return { status: 'OK' };
};

export const handler = new Corva().scheduled(scheduledApp);

Contributing

Set up the project

$ cd ~/YOUR_PATH/node-sdk
$ npm ci

Run tests

$ npm test

Run code linter

$ npm run lint