AsyncQueue with progress?

[airtonix]

Looking at using pacer server side, I realised we'd have to facilitate our own implentation for tracking task progress while in the queue.

Is this something people would want in this lib or as a separate pkg?

import EventEmitter from 'events';

import { AsyncQueuer, type AsyncQueuerOptions } from '@tanstack/pacer';
import { uniqueId } from 'lodash-es';

type Task<T> = { id: string; result: T };
type ProgressTask<T> = (state: TaskState<T>) => Promise<T>;
type TaskState<T> = {
  id: string;
  progress: {
    value: number;
    forward(value: number, state?: T): void;
  };
};
enum ProgressDirection {
  /**
   * Progress moves from 0 to 100
   */
  Ascending,
  /**
   * Progress moves from 100 to 0
   */
  Descending,
}

type EventedProgress = number;
type EventedId = string;
class EventedProgressMap<T> extends EventEmitter<{
  start: [EventedId, EventedProgress];
  progress: [EventedId, EventedProgress, T | undefined];
  completed: [EventedId, T | undefined];
}> {
  store = new Map<EventedId, EventedProgress>();

  constructor(
    public direction: ProgressDirection = ProgressDirection.Descending,
  ) {
    super();
  }

  get(id: EventedId) {
    return this.store.get(id);
  }

  set(id: EventedId, value: EventedProgress, result?: T) {
    this.store.set(id, value);
    this.emit('progress', id, value, result);
  }

  forward(id: EventedId, value: EventedProgress, state?: T) {
    const current = this.get(id);
    if (!current) {
      throw new Error(`No Progress Item with id ${id}`);
    }

    if (
      this.direction === ProgressDirection.Ascending &&
      current + value >= 100
    ) {
      this.set(id, 100, state);
      return;
    }

    if (this.direction === ProgressDirection.Ascending) {
      this.set(id, current + value, state);
      return;
    }

    if (
      this.direction === ProgressDirection.Descending &&
      current - value <= 0
    ) {
      this.set(id, 0, state);
      return;
    }

    if (this.direction === ProgressDirection.Descending) {
      this.set(id, current - value, state);
      return;
    }
  }

  start(id: EventedId) {
    const value = this.direction === ProgressDirection.Ascending ? 0 : 100;
    this.store.set(id, value);
    this.emit('start', id, value);
  }

  completed(id: EventedId, result: T) {
    this.store.delete(id);
    this.emit('completed', id, result);
  }
}

class AsyncProgressQueuer<T> extends AsyncQueuer<Task<T>> {
  progress: EventedProgressMap<T>;

  constructor(
    options?: AsyncQueuerOptions<Task<T>> & { progress: ProgressDirection },
  ) {
    super(options);
    this.progress = new EventedProgressMap(
      options?.progress || ProgressDirection.Descending,
    );
  }

  createTaskState(id: string) {
    const get = () => this.progress.get(id) || 0;
    const set = (value: number, state?: T) =>
      this.progress.set(id, value, state);

    const progress: TaskState<T>['progress'] = {
      get value() {
        return get();
      },
      set value(value: number) {
        set(value);
      },
      forward: (value, state) => {
        this.progress.forward(id, value, state);
      },
    };

    const state: TaskState<T> = {
      id,
      progress,
    };

    return state;
  }

  addTask(fn: ProgressTask<T>) {
    const id = uniqueId('task');

    const task = async () => {
      this.progress.start(id);
      const result = await fn(this.createTaskState(id));
      this.progress.completed(id, result);

      // return a wrapped object that helps us identify the task on success
      return {
        id,
        result,
      };
    };

    return this.addItem(task);
  }
}

usage

const pause = async (ms: number) =>
  await new Promise((resolve) => setTimeout(resolve, ms));

type Channel = {
  channel: number;
  meta?: Record<string, string>;
};

const ChannelQueue = new AsyncProgressQueuer<Channel>();
ChannelQueue.progress.on('start', (id, progress) => {
  console.log('[QUEUE] item started', id, progress);
});
ChannelQueue.progress.on('progress', (id, progress, state) => {
  console.log('[QUEUE] item progressed', id, progress, state);
});
ChannelQueue.progress.on('completed', (id, result) => {
  console.log('[QUEUE] item completed', id, result);
});

ChannelQueue.addTask(async ({ progress }) => {
  const data: Channel = { channel: 4 };
  await pause(1000);
  progress.forward(33, { ...data, meta: { tone: 'puppies' } });
  await pause(1000);
  progress.forward(33);
  await pause(1000);
  progress.forward(33);
  return Promise.resolve({ channel: 4 });
});

ChannelQueue.addTask(async ({ progress }) => {
  progress.forward(50);
  await pause(1000);
  progress.forward(50);
  return Promise.resolve({ channel: 34 });
});

// Monitor task states
console.log('activeItems', ChannelQueue.getActiveItems()); // Currently processing tasks
console.log('pendingItems', ChannelQueue.getPendingItems()); // Tasks waiting to be processed

ChannelQueue.start();

results in

tsx ./apps/manager/app/core/queue.ts
[QUEUE] item started task1 100
activeItems [ [AsyncFunction (anonymous)] { priority: undefined } ]
pendingItems [ [AsyncFunction (anonymous)] { priority: undefined } ]
[QUEUE] item progressed task1 67 { channel: 4, meta: { tone: 'puppies' } }
[QUEUE] item progressed task1 34 undefined
[QUEUE] item progressed task1 1 undefined
[QUEUE] item completed task1 { channel: 4 }
[QUEUE] item started task2 100
[QUEUE] item progressed task2 50 undefined
[QUEUE] item progressed task2 0 undefined
[QUEUE] item completed task2 { channel: 34 }

[KevinVandy]

So making a @tanstack/server-pacer package was definitely one of my original goals, but... so much would have to go into making that correct, and also potentially documenting or making adapters for state persistence with Reddis key value-like stores or durable objects, that I'm personally focussing on the client-side functionality primitives first.
Happy to discuss possiblities of a server-pacer package. Maybe certain aspects would be much simpler than I'm thinking.

[airtonix]

Glad to start the process of awareness.

Me personally... I'm probably fairly ignorant of all the things that need to be accounted for.

Interrogation of an inflight/pending tasks is definitely one of my wishlist items.