# songwhip-queue

A service to schedule (one or many) HTTP requests, with smart back-off retry logic.

- Used in Songwhip to call untrusted third-party APIs that may fail due to rate-limits or traffic spikes.
- Two APIs: Workflows and Groups
  - Worklows are composed of Steps and are ideal to run a small number of HTTP requests that might depend on each other (i.e. Request2 needs the output of Request1 to run). Workflows are not persisted anywhere and thus provide very limited auditing capabilities.
  - Groups are composed of Tasks and are ideal to run large number of HTTP requests that don't depend on each other. Groups and Tasks are persisted in DynamoDB and thus are able to be inspected during/after the run.

## Architecture

```mermaid
flowchart TB
    Client(("Client"))
    CreateWorkflowHandler["Create Workflow Handler"]
    SQS[("SQS Queue")]
    Lambda["Lambda"]
    RunWorkflowHandler["Run Workflow Handler"]

    Client -- POST https://queue.songwhip.com --> CreateWorkflowHandler
    CreateWorkflowHandler --> SQS
    SQS --> Lambda
    Lambda -- POST https://queue.songwhip.com/sqs/webhook --> RunWorkflowHandler

    subgraph Vercel
    CreateWorkflowHandler
    RunWorkflowHandler
    end

    subgraph AWS
    SQS
    Lambda
    end

    Client ~~~ CreateWorkflowHandler
```

## Examples

### Workflows: Create Appreciation Engine Brand/Segment

Here songwhip-queue is used from songwhip-api to safely setup Appreciation Engine Brand/Segment objects for a Songwhip Album page and callback when completed. Upon callback songwhip-api can store the Appreciation Engine references so that when a fan presaves the fan data can be pushed into the correct Segment.

```ts
await fetch('https://queue.songwhip.com/', {
  method: 'POST',

  body: JSON.stringify({
    steps: [
      {
        url: 'https://appreciation-engine.songwhip.com/segments',
        method: 'PUT',

        body: {
          artistName: "Name",
          songwhipAlbumId: 111,
          …
        }
      },
      {
        url: 'https://api.songwhip.com/webooks/appreciation-engine/onSegmentCreated',
        method: 'POST',

        body: {
          segmentId: '${steps[0].response.body.data.segment.ID}',
          brandId: '${steps[0].response.body.data.brand.ID}',
          albumId: 111,
        }
      }
    ],
  })
});
```

### Workflows: Push fan data into Appreciation Engine

This example shows songwhip-queue could be used from songwhip-api to safely push the fan profile and refresh token into Appreciation Engine and then send them an email once it's succeeds.

```ts
await fetch('https://queue.songwhip.com/', {
  method: 'POST',

  body: JSON.stringify({
    steps: [
      {
        url: 'https://appreciation-engine.songwhip.com/segments/SEGMENT_ID/members',
        method: 'PUT',

        body: {
          "email": "wilson@example.com",
          "serviceType": "spotify",
          "serviceUserToken": "SECRET_REFRESH_TOKEN",
          "serviceUserId": "wilsonpage"
        }
      },
       {
        url: 'https://email.songwhip.com/send',
        method: 'PUT',

        body: {
          "subject": "Thanks for presaving 🎉"
          "bodyMarkdown": "Thanks for presaving",
          "fromEmail": "contact@songwhip.com",
          "fromName": "Songwhip",
          "toEmails": ["wilson@example.com"],
        }
      }
    ],
  })
});
```

### Groups: Execute a large number presave tasks

Once a release goes live we need to execute all the presaves that have been stored in songwhip-release-tasks. Here songwhip-release-tasks can pull all the tasks for a particular album and leverage songwhip-queue to execute the tasks reliably. We can split the tasks by type (e.g. Spotify presaves, Apple Music presaves, ...) and create a Group for each type. Using multiple Groups here will allow us to keep running Apple Music presave tasks even if a rate-limit with the Spotify API has been reached for example.

```ts
// Create a Group for the Spotify presaves
await fetch('https://queue.songwhip.com/groups/album1-spotify-presave', {
  method: 'PUT',

  body: JSON.stringify({
    // Group-level options that will apply to all Tasks.
    // The same options are available at the Task-level
    taskDefaults: {
      // Tasks that fail with a 400 or a 404 will no be retried and will be skipped,
      // allowing the next Tasks to be run
      noRetryOnStatus: [400, 404],

      // Every Task that fail with a 429 or a 502 should not be skipped and will be
      // retried until it succeeds. Until then no other Tasks will be run.
      skipOnError: {
        not: [429, 502],
      },
    },

    tasks: [
      {
        url: 'https://release-tasks.songwhip.com/album1/task1/run',
        method: 'POST',
      },
      {
        url: 'https://release-tasks.songwhip.com/album1/task2/run',
        method: 'POST',
      },
      {
        url: 'https://release-tasks.songwhip.com/album1/task3/run',
        method: 'POST',
      },
      ...
    ],
  }),
});

// Create a Group for the Apple Music presaves
await fetch('https://queue.songwhip.com/groups/album1-apple-music-presave', {
  method: 'PUT',

  body: JSON.stringify({
    taskDefaults: {
      ...
    },

    tasks: [
      {
        url: 'https://release-tasks.songwhip.com/album1/task11/run',
        method: 'POST',
      },
      {
        url: 'https://release-tasks.songwhip.com/album1/task22/run',
        method: 'POST',
      },
      {
        url: 'https://release-tasks.songwhip.com/album1/task33/run',
        method: 'POST',
      },
      ...
    ],
  }),
});
```

## Development

### Prerequisites

- Node.js configured repo\
  https://www.notion.so/Node-js-project-setup-guide-86feadffeb304962b06e6b1bb8f09213
- Docker
- Vercel CLI

### Running

- Create a `.env` file based on the `.env.shadow` template
- Run `yarn dev` to spin up a local dev server and a local dockerized instance of Localstack to emulate AWS services

### Testing

- Run `yarn test` to check for format, lint or types errors and run the test suites
- You can also use `yarn test:watch` to run the test suites in watch mode.

### Deploying the Lambda proxy

The Lambda proxy just forwards requests to the Vercel deployment so it should rarely change. In the event you do need to update it you can do so manually from you local machine.

1. [Setup `awsume` on you local machine](https://www.notion.so/AWS-Access-f841b9dd815d4443a80e96a86c92cd2f#309ab52b8f254d37bf8660110b2c2e86) to be able to reach the songwhip aws accounts using AWS CLi.
2. `awsume songwhip-{prod|qa}`
3. `yarn deploy:updateLambda {prod|qa}-lambda-songwhip-queue`

### Metrics and Telemetry

This project uses a custom telemetry system to send metrics to Datadog, helping monitor the performance and behavior of our task queue system.

#### Collected Metrics

| Metric Name         | Type  | Description                                                    | Tags                                                              |
| ------------------- | ----- | -------------------------------------------------------------- | ----------------------------------------------------------------- |
| group.started       | Count | Emitted when a group starts processing                         | group_id                                                          |
| group.stopped       | Count | Emitted when a group stops processing                          | group_id, task_id, reason, error_message, error_status            |
| group.recovered     | Count | Emitted when a stopped or queued group is re-driven via /start | group_id, from_status                                             |
| task.started        | Count | Emitted when a task starts                                     | group_id, task_id, reason, error_message, error_status            |
| task.completed      | Count | Emitted when a task completes (success, error, or skipped)     | group_id, task_id, status, duration, error_status (if applicable) |
| pending_tasks_group | Gauge | Remaining pending tasks per group (total - completed - failed) | group_id                                                          |

#### Metric Details

- `group.started`: Tracks the initiation of group processing.
- `group.stopped`: Indicates when a group stops processing, with reasons and error status if applicable.
- `group.recovered`: Emitted when a `stopped` or `queued` group is manually re-driven via `POST /groups/:groupId/start`. The `from_status` tag records the status the group was recovered from.
- `task.started`: Marks the beginning of individual task execution.
- `task.completed`: Comprehensive metric for task completion, covering all outcomes (success, error, skipped).
- `songwhip.queue.service.pending_tasks_group`: Track's pending tasks per group, updating as tasks complete or fail. Used for monitoring group processing progress;

#### Configuration

Telemetry is configured using the following environment variables:

- `DD_SERVICE`: Service name (default: 'songwhip-queue')
- `DD_ENV`: Environment (e.g., 'production', 'staging')
- `DD_VERSION`: Application version (commit SHA is used by default)
- `DD_API_KEY`: Datadog API key
- `DD_TRACES_ENDPOINT`: Endpoint for sending traces
- `DD_METRICS_ENDPOINT`: Endpoint for sending metrics
- `DD_ALLOW_SEND_METRICS`: Allow sending metrics
- `DD_ALLOW_SEND_TRACES`: Allow sending traces

#### Monitoring and Alerting

Use these metrics in Datadog to:

- Create dashboards visualizing queue performance
- Set up alerts for abnormal behavior (e.g., high error rates, long-running tasks)
- Track performance over time

Adjust Datadog settings to properly collect and display these custom metrics