simple-async-tasks

simple-async-tasks

easily create and use async-tasks within a pit-of-success

install

npm install simple-async-tasks

use

define your async task

define the domain object of the async task you want to be able to run

import { DomainEntity } from 'domain-objects';
import { AsyncTask, AsyncTaskStatus } from 'simple-async-tasks';

/**
 * for example: an async task for emitting some data to remote persistance
 */
export interface AsyncTaskEmitToRemote extends AsyncTask {
  uuid?: string;
  updatedAt?: string;
  status: AsyncTaskStatus;

  /**
   * the endpoint to emit the data to
   */
  endpoint: string;

  /**
   * the payload to emit
   *
   * note
   * - supports string and binary buffer
   */
  payload: string | Buffer;
}
export class AsyncTaskEmitToRemote
  extends DomainEntity<AsyncTaskEmitToRemote>
  implements AsyncTaskEmitToRemote
{
  public static unique = ['endpoint', 'payload'];
}

define your dao

define the database-access-object we can use to persist this async-task

usually, you should be using a library to code-generate or instantiate the dao for you

• e.g., sql-dao-generator
• e.g., dynamodb-dao-generator
• e.g., cache-dao-generator

for example

import { createCacheDao } from 'simple-cache-dao';
import { createCache } from 'simple-in-memory-cache';

const daoTaskEmitToRemote = createCacheDao({ cache: createCache() })

define how to queue

define how to queue your task for execution

import { createQueue, QueueOrder } from 'simple-in-memory-queue';

// TODO: load all queued tasks from db on page load
export const asyncTaskEmitToRemoteQueue = createQueue<AsyncTaskEmitToRemote>({
  order: QueueOrder.FIRST_IN_FIRST_OUT,
});

export const queueTaskEmitToRemote = withAsyncTaskExecutionLifecycleQueue({
  dao: daoTaskEmitToRemote,
  queue: asyncTaskEmitToRemoteQueue,
  getNew: ({ endpoint, payload }) =>
    new AsyncTaskEmitToRemote({
      status: AsyncTaskStatus.QUEUED,
      endpoint,
      payload,
    }),
});

define how to execute

define how to execute your async task

export const executeTaskEmitToRemote = withAsyncTaskExecutionLifecycleExecute(
  async ({ task }: { task: HasMetadata<AsyncTaskEmitToRemote> }) => {
    // execute your logic

    // mark it as fulfilled
    await daoTaskEmitToRemote.upsert({ task: { ...task, status: AsyncTaskStatus.FULFILLED }})
  },
  {
    dao: daoTaskEmitToRemote,
  },
);

⚠️ note: you must change the status of the task away from attempted by the time the execute function resolves to some non-attempted , otherwise it will be considered a failure

define the execution trigger

define the trigger that will consume from your queue and invoke the execute function

note

• this will vary based on which queue implementation you use
• for example,
  • if using aws.sqs, you will probably invoke an aws.lambda with an aws.event-source-mapping
  • if using a simple-in-memory-queue, you may choose to use a resilient remote consumer

Changelog

1.8.0 (2024-08-19)

Features

• extract: simplify extracting tasks from sqs events (75a6891)
• init: initialize based on simple-localstorage-cache (dd78966)
• mutex: support mutex lock retries via dobj class statics (23b923e)
• observability: add logs. only warn on redundant attempts (#7) (45f74c6)
• requeue: add native sqs requeue capability to prevent error logs on retryLater (ca83c9b)
• sqs: distribute terraform queue module (f33a02c)
• sqs: expose utility to get sqs queue url from task name (a9f2335)

Bug Fixes

• deploy: bump to overcome ghost npm version (db154a7)
• deploy: bump to overcome ghost release version (83293d6)
• deps: remove unused dep (87b6db1)
• deps: unpin typefns peer dep with new v sempher range (#14) (87af9f8)
• deps: unpin typefns peer dependency (#12) (acbdece)
• dobj: support updatedAt as Date or iso String (1bd381d)
• enqueue: re-enable enqueue to any queue, beyond sqs (495141c)
• enqueue: support input type which extends unique key (816f060)
• exec: auto-retry on unfound task for read-after-write resilience (597b2b8)
• execute: enable custom attempt timeout in seconds (af67c8c)
• logs: improve log observability (48a6bbc)
• mutex: ensure mutex blocks only upto max lambda duration (62dca67)
• pkg: make type-fns a peer dep (b417a50)
• practs: bump practs to latest best (05d6df8)
• queue: allow queue url defined via async fn (c8ecfe1)
• retry: ensure to exit without attempt on retry later via requeue (653ec01)
• sqs: ensure sqs meta driven retries prevent infiloops (34fb491)
• types: support procedure.input,.context args pattern (0204a8b)