Class Corva

Scheduled app

Scheduled apps can be based on:

• data time - invoked each <configured interval> seconds, while data available
• depth time - invoked each <configured interval> feet
• natural time - invoked each <configured interval> seconds

Example

TypeScript

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

export const handler = new Corva().scheduled(
  async (event: ScheduledDataTimeEvent, { api, logger }) => {
    const { body: records } = await api.request<CollectionRecord<unknown>>('api/v1/data/my-provider/my-collection/', {
      method: 'GET',
      searchParams: {
        query: JSON.stringify({
          asset_id: event.asset_id,
          timestamp: {
            $gte: event.start_time,
            $lte: event.end_time,
          },
        }),
        sort: JSON.stringify({ timestamp: 1 }),
        limit: 100,
        skip: 0,
      },
    });

    logger.info(records);

    return { status: 'OK' };
  },
  {
    updateStatus: true,
    swallowErrors: true,
  },
);

Example

JavaScript

const { Corva } = require('@corva/node-sdk');

exports.handler = new Corva().scheduled(async (event, { logger, api }) => {
  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: {
          $gte: event.start_time,
          $lte: event.end_time,
        },
      }),
      sort: JSON.stringify({ timestamp: 1 }),
      limit: 100,
      skip: 0,
    },
  });

  logger.info(records);

  return { status: 'OK' };
});

Realtime app

Stream apps can be time (need to rely on timestamp) or depth (need to rely on measured_depth) based.

Example

TypeScript

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

type Drillstring = {
  bit_depth: number;
  gamma_ray: number;
  components: {
    gamma_sensor_to_bit_distance: number;
  }[];
};

export const handler = new Corva().stream(
  async (event: StreamTimeEvent<Drillstring>, { api }) => {
    // do some calculations/modifications to the data
    const total = event.records.reduce((acc, drillstring) => acc + drillstring.data.bit_depth, 0);
    const average = total / event.records.length;

    // save the data to private collection
    await api.request('api/v1/data/my-provider/drillstrings.average-invokes/', {
      method: 'POST',
      json: [
        {
          timestamp: Date.now() / 1000,
          company_id: 42,
          asset_id: event.asset_id,
          version: 1,
          data: { average },
        },
      ],
    });
  },
  {
    filteringMode: FilteringMode.Timestamp,
    swallowErrors: false,
  },
);

Example

JavaScript

const { Corva, FilteringMode } = require('@corva/node-sdk');

const streamApp = async (event, { api }) => {
  // get some data from api
  const { body: drillstrings } = await api.request('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, {
  filteringModes: FilteringMode.TIMESTAMP,
});

Task app

Example

TypeScript

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

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

interface DatasetRecord {
  // eslint-disable-next-line camelcase
  data: { some_valuable_prop: number };
}

/**
 *  This will appear in {@link Task.payload}
 */
interface TaskOutput {
  opts: string;
  some: {
    output: number;
  };
}

const timezonesShift: Record<string, number> = {
  'America/Chicago': 6,
};

export const handler = new Corva().task<TaskProps, void>(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.some_valuable_prop'],
  });

  // do some calculations
  const sum = results.reduce((acc, item) => acc + item.data.some_valuable_prop, 0);

  // save the results somewhere
  await api
    .provider('some-provider')
    .dataset<TaskOutput>('some-dataset')
    .createEntry(
      {
        opts: event.properties.some.input,
        some: { output: sum },
      },
      { timestamp: Date.now() },
    );
});

Example

JavaScript

/* eslint-disable camelcase */
const { Corva } = require('@corva/node-sdk');

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

const taskApp = 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({
    dataset: 'some-dataset',
    query: {
      timestamp: { $gte: timestampToLoadDataFrom },
    },
    sort: { timestamp: 1 },
    limit: 1000,
    fields: ['data.some_valuable_prop'],
  });

  // do some calculations
  const sum = results.reduce((acc, item) => acc + item.data.some_valuable_prop, 0);

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

exports.handler = new Corva().task(taskApp);