Build a Work Queue Slack App with TypeScript

export interface WorkqueueData {
  id: string;
  timestamp: string;
  channelName: string;
  channelId: string;
  userId: string;
  work: string;
  status: WorkqueueStatus;
  claimantId?: string;
  // Add more properties as needed
}

export enum WorkqueueStatus {
  Backlog = 1,
  InProgress = 2,
  Done = 3,
}

import {
  condition,
  continueAsNew,
  isCancellation,
  workflowInfo,
  // ...
} from '@temporalio/workflow';
import { WorkqueueData } from '../../../common-types/types';
// ...
export async function workqueue(existingData?: WorkqueueData[]): Promise<void> {
  const wqdata: WorkqueueData[] = existingData ?? [];
  // ...
  try {
    // Await until suggestion to Continue-As-New due to History size
    // If a Cancellation request exists, the condition call will throw the Cancellation error
    await condition(() => workflowInfo().continueAsNewSuggested);
  } catch (e) {
    // Catch a Cancellation error
    if (isCancellation(e)) {
      // Set the Workflow status to Cancelled by throwing the CancelledFailure error
      throw e;
    } else {
      // Handle other types of errors
      throw e;
    }
  }
  await continueAsNew<typeof workqueue>(wqdata);
}

import {
  // ...
  defineQuery,
  defineSignal,
  setHandler,
} from '@temporalio/workflow';
// ...
export const getWorkqueueDataQuery = defineQuery<WorkqueueData[]>(
  'getWorkqueueData',
);
export const addWorkToQueueSignal = defineSignal<[WorkqueueData]>(
  'addWorkqueueData',
);
export const claimWorkSignal = defineSignal<
  [{ workId: string; claimantId: string }]
>('claimWork');
export const completeWorkSignal = defineSignal<[{ workId: string }]>(
  'completeWork',
);

export async function workqueue(existingData?: WorkqueueData[]): Promise<void> {
  // ...
  // Register a Query handler for 'getWorkqueueData'
  setHandler(getWorkqueueDataQuery, () => {
    return wqdata;
  });

  // Register the Signal handler for adding work
  setHandler(addWorkToQueueSignal, (data: WorkqueueData) => {
    wqdata.push(data);
  });

  // Register Signal handler for claiming work
  setHandler(claimWorkSignal, ({ workId, claimantId }) => {
    const workItem = wqdata.find((item) => item.id === workId);
    if (workItem) {
      workItem.claimantId = claimantId;
      workItem.status = 2;
    }
  });

  // Register Signal handler for completing work
  setHandler(completeWorkSignal, ({ workId }) => {
    const index = wqdata.findIndex((item) => item.id === workId);
    if (index !== -1) {
      wqdata.splice(index, 1);
    }
  });
  // ...
}

import 'dotenv/config';
import { NativeConnection, Worker } from '@temporalio/worker';
import path from 'path';

async function run() {
  try {
    const worker = await Worker.create({
      namespace: process.env.TEMPORAL_DEV_NAMESPACE || '',
      workflowsPath: path.resolve(__dirname, './workflows'),
      taskQueue: `${process.env.ENV}-temporal-iq-task-queue`,
    });

    await worker.run();
  } catch (err) {
    console.error(err);
    process.exit(1);
  }
}

run();

import 'dotenv/config';
import { Client, Connection } from '@temporalio/client';

export let temporalClient: Client;

export async function initializeTemporalClient() {
  const connection = await Connection.connect();

  temporalClient = new Client({
    connection,
    namespace: process.env.TEMPORAL_DEV_NAMESPACE!,
  });
}

import 'dotenv/config';
import {
  App,
  // ...
} from '@slack/bolt';
import { initializeTemporalClient } from './modules/dev-temporal-client';
// ...
// Initializes your app with your bot token, app token, and signing secret
const app = new App({
  token: process.env.SLACK_BOT_TOKEN!,
  signingSecret: process.env.SLACK_SIGNING_SECRET!,
  socketMode: true,
  appToken: process.env.SLACK_APP_TOKEN!,
});
// ...
// Register Slack bot error handler
app.error(async ({ error }: { error: Error }) => {
  if (error instanceof Error) {
    console.error(`Error: ${error.name}, Message: ${error.message}`);
  } else {
    console.error('An unknown error occurred', error);
  }
});

// Start the app
(async () => {
  try {
    await app.start();
    await initializeTemporalClient();
    console.log('⚡️ Bolt app is running!');
  } catch (error) {
    console.error('Failed to start Bolt app:', error);
  }
})();

// ...
import {
  RespondFn,
  SayFn,
  SlackCommandMiddlewareArgs,
  // ...
} from '@slack/bolt';
import { WorkqueueData, WorkqueueStatus } from '../../common-types/types';
import { temporalClient } from './dev-temporal-client';
// ...
import { WorkflowExecutionAlreadyStartedError } from '@temporalio/client';
// ...
// Handles and routes all incoming Work Queue Slash Commands
export async function handleWorkqueueCommand(
  command: SlackCommandMiddlewareArgs['command'],
  say: SayFn,
  respond: RespondFn,
) {
  const commandText = command.text?.trim();

  if (commandText === '!delete') {
    await deleteWorkqueue(command, say);
  } else if (commandText === '') {
    await displayWorkQueue(command, respond);
  } else {
    await addWorkToQueue(command, say);
  }
  return;
}

// ...
// Display the Work Queue for the channel
// Creates a new Work Queue if it does not exist
async function displayWorkQueue(
  command: SlackCommandMiddlewareArgs['command'],
  respond: RespondFn,
) {
  // Get the channel name in plain text
  const channelName = command.channel_name;
  // Create a new Work Queue for the channel
  await createNewWorkQueue(channelName);
  // If the Work Queue already exists, Query it
  const data = await queryWorkQueue(channelName, respond);
  await replyEphemeral(
    respond,
    'Work Queue cannot display',
    formatWorkqueueDataForSlack(channelName, data),
  );
}

// Create a new Work Queue for the channel if one does not exist
async function createNewWorkQueue(workflowid: string): Promise<void> {
  try {
    await temporalClient.workflow.start('workqueue', {
      taskQueue: `${process.env.ENV}-temporal-iq-task-queue`,
      workflowId: workflowid,
    });
  } catch (e) {
    if (e instanceof WorkflowExecutionAlreadyStartedError) {
      console.log('Workflow already started');
    } else {
      throw e;
    }
  }
}

// Read the state of the Work Queue for the channel using a Query
async function queryWorkQueue(
  workflowId: string,
  say: SayFn,
): Promise<WorkqueueData[]> {
  try {
    const handle = temporalClient.workflow.getHandle(workflowId);
    const result = await handle.query<WorkqueueData[]>(getWorkqueueDataQuery);
    console.log('Current workqueue data:', result);
    return result;
  } catch (error) {
    console.error('Error querying workqueue data:', error);
    await say('An error occurred while Querying the Work Queue.');
    return [];
  }
}

// ...
// Add work to the queue using a Signal
async function addWorkToQueue(
  command: SlackCommandMiddlewareArgs['command'],
  say: SayFn,
) {
  // Get the channel name in plain text
  const channelId = command.channel_id;
  const channelName = command.channel_name;
  const wqdata = buildWQData(command, channelId, channelName);
  await signalAddWork(wqdata, say);
  // Reply to the message directly in the thread
  await reply(say, `Added Work ${wqdata.id} to the Queue.`);
}
// ...
async function signalAddWork(params: WorkqueueData, say: SayFn): Promise<void> {
  try {
    await temporalClient.workflow.signalWithStart('workqueue', {
      workflowId: params.channelName,
      taskQueue: `${process.env.ENV}-temporal-iq-task-queue`,
      signal: addWorkToQueueSignal,
      signalArgs: [params],
    });
  } catch (error) {
    console.error('Error signaling workqueue data:', error);
    await say('An error occurred while Signaling the Work Queue.');
  }
}

// ...
// Listen for Work Item Claim
app.action<BlockAction<BlockElementAction>>(
  'wq_claim',
  async ({ ack, say, body }) => {
    await ack();
    // Ensure the body.actions[0] is a ButtonAction
    const action = body.actions[0] as ButtonAction;
    if (action.value) {
      const [channelName, workId, userId] = action.value.split('_');
      const claimantId = body.user.id;
      // Send signal to the Temporal workflow to claim the work
      await signalClaimWork(channelName, workId, claimantId, userId, say);
    } else {
      console.error('Action value is undefined.');
    }
  },
);

// ...
export async function signalClaimWork(
  channelName: string,
  workId: string,
  claimantId: string,
  userId: string,
  say: SayFn,
) {
  try {
    const handle = temporalClient.workflow.getHandle(channelName);
    await handle.signal(claimWorkSignal, { workId, claimantId });
    console.log(`Work item ${workId} claimed by ${claimantId}`);
    await reply(
      say,
      `<@${userId}> Work item ${workId} claimed by <@${claimantId}>.`,
    );
  } catch (error) {
    console.error('Failed to signal claim work:', error);
  }
}

// ...
// Listen for Work Item Completion
app.action<BlockAction<BlockElementAction>>(
  'wq_complete',
  async ({ ack, say, body }) => {
    await ack();
    const action = body.actions[0] as ButtonAction;
    if (action.value) {
      const [channelName, workId, userId] = action.value.split('_');
      const message = body.message as GenericMessageEvent;
      // Send signal to the Temporal workflow to complete the work
      await signalCompleteWork(channelName, workId, message, userId, say);
    } else {
      console.error('Action value is undefined.');
    }
  },
);

// ...
export async function signalCompleteWork(
  channelId: string,
  workId: string,
  message: GenericMessageEvent,
  userId: string,
  say: SayFn,
) {
  try {
    const handle = temporalClient.workflow.getHandle(channelId);
    await handle.signal(completeWorkSignal, { workId });
    console.log(`Work item ${workId} marked as complete`);
    await reply(say, `<@${userId}> Work item ${workId} marked as complete.`);
  } catch (error) {
    console.error('Failed to signal complete work:', error);
  }
}

// ...
// Delete the Work Queue for the channel with a Cancellation Request
export async function deleteWorkqueue(
  command: SlackCommandMiddlewareArgs['command'],
  say: SayFn,
): Promise<void> {
  const workflowId = command.channel_name;
  try {
    const handle = temporalClient.workflow.getHandle(workflowId);
    await handle.cancel();
    console.log(`Workflow with ID ${workflowId} has been cancelled.`);
    await reply(say, `Work Queue has been deleted for this channel.`);
  } catch (error) {
    console.error(`Failed to cancel workflow with ID ${workflowId}:`, error);
    await reply(say, `Failed to delete Work Queue for this channel.`);
  }
}

import { WorkflowCoverage } from '@temporalio/nyc-test-coverage';
import { TestWorkflowEnvironment } from '@temporalio/testing';
import { DefaultLogger, Runtime, Worker } from '@temporalio/worker';
// ...
describe('Work Queue Workflow', () => {
  let testEnv: TestWorkflowEnvironment;
  const workflowCoverage = new WorkflowCoverage();

  beforeAll(async () => {
    Runtime.install({ logger: new DefaultLogger('WARN') });
    testEnv = await TestWorkflowEnvironment.createLocal();
  });

  afterAll(async () => {
    await testEnv?.teardown();
  });

  afterAll(() => {
    workflowCoverage.mergeIntoGlobalCoverage();
  });
  // ...
});

// ...
test('should add work to the queue', async () => {
  // Get a test environment Temporal Client
  const { client, nativeConnection } = testEnv;
  // Create a test environment Worker
  const worker = await Worker.create(
    workflowCoverage.augmentWorkerOptions({
      connection: nativeConnection,
      taskQueue: 'test',
      workflowsPath: require.resolve('../workflows'),
    }),
  );
  // ...
  // Run the Worker
  await worker.runUntil(async () => {
    const handle = await client.workflow.start(workqueue, {
      args: [],
      workflowId: workflowId,
      taskQueue: 'test',
    });
    const workItem: WorkqueueData = {
      // ...
    };
    // Add work to the queue
    await handle.signal(addWorkToQueueSignal, workItem);
    // Check to see if the data is there
    const result = await handle.query(getWorkqueueDataQuery);
    // Compare the data
    expect(result).toContainEqual(workItem);
  });
});