How to use Firebase queues

If you’re a Firebase user you probably appreciate the many built-in capabilities like real-time snapshots, Firestore triggers, and authentication. I was just learning about Deno Queues and realized after years of using Firebase I hadn’t really read about Firebase Queues. Just Where Are Firebase Lines?

Queue Use Case for Firebase

Ayrshare is built on top of Firebase, making heavy use of Firebase Services. Ayrshare is a social media API and we allow users to publish or receive statistics on social media. However, as we grew, we realized that the way we were processing scheduled posts was wrong and exceeded the limits of the architecture.

Whenever a user schedules a post using our public API, we create an entry in the Firestore collection – a home grown queue. A Firebase scheduled job runs once a minute to process all pending posts. This means that multiple scheduled posts are processed together, with the exception of posts that may affect other posts, and the job has a timeout of nine minutes per Firebase limit.

An obvious improvement is that the scheduled job processes one post at a time, but that can mean parallel processing which can significantly slow down publishing. Although we can overcome these challenges with more courage, in the end we will be building our own complex line. What we really need is a way to queue jobs and process them independently.

Equeue Tasks With Cloud Tasks

After many searches along the lines of “Firebase” you will come across something called “Enqueue Functions with Cloud Tasks”. As Firebase explains:

Task queue tasks use Google Cloud Tasks to help your application run time-consuming, resource-intensive, or bandwidth-intensive tasks concurrently, outside of your main application’s flow.

At first glance this may not sound like the right solution, but dig deeper and you’ll see that it’s the perfect answer, if a little complicated.

To put it in simple terms, you create a Firebase queue, i.e. a Cloud Task, add data to the queue, and another Firebase process will run on each item, i.e. a task, in the queue.

How to use Firebase Queue

Let’s dive into some code on how to create, add jobs, and process those jobs in Firebase.

Create and add to the queue

We’ll start by creating a queue and adding jobs.

const { getFunctions } = require("firebase-admin/functions");
const enqueueScheduledPosts = async () => {
  log("Starting enqueueScheduledPosts");

  const posts = await gatherScheduledPosts();

  const queue = getFunctions().taskQueue("postTask");
  const targetUri = await getFunctionUrl("postTaskFunction");

  const enqueues = [];
  posts.forEach((post) => {
    enqueues.push(
      queue.enqueue(post, {
        scheduleDelaySeconds: 1,
        dispatchDeadlineSeconds: 60 * 8, // 8 minutes
        uri: targetUri
      })
    );
  });

  await Promise.all(enqueues).catch((err) =>
    el("Error with enqueues:", posts, err)
  );

  log("Ending enqueueScheduledPosts");
};

This is the first job gatherScheduledPosts gathers all the spaces that need to be processed.

We then create a named row postTask with getFunctions().taskQueue("postTask"). All posts will be added, indented, to this line.

Next, we get the targetUri of work with getFunctionUrl("postTaskFunction") which will process each job in line. Locating the URI is discussed below.

Now we can queue each function queue.enqueue(data, options). You pass in parameters data that contain the information you want to process and the JSON string options. In our example we send:

• scheduleDelaySeconds – How much delay there is between each processing operation. This is optional.
• dispatchDeadlineSeconds – Timeout for processing jobs. This is optional.
• uri – The target URI of the job to process jobs. This is necessary.

I enqueue function returns a promise, so you can process them with the Promise.all function. Once completed, all tasks are queued and ready to run.

Get the Queue Task URI

The function URI is simply the URL of the function that will be called by each function:

While you can hard-code the URI of the job processing function, it’s best to reuse it and take care to do the job lookup.

const getFunctionUrl = async (name, location = "us-central1") => {
  const { GoogleAuth } = require("google-auth-library");
  const gAuth = new GoogleAuth({
    scopes: "
  });
  const projectId = await gAuth.getProjectId();

  const url =
    " +
    `projects/${projectId}/locations/${location}/functions/${name}`;

  const client = await gAuth.getClient();
  const res = await client.request({ url });
  const uri = res.data?.serviceConfig?.uri;
  if (!uri) {
    throw new Error(`Unable to retreive uri for function at ${url}`);
  }

  console.log("Function URL for Task:", name, uri);

  return uri;
};

All of this code looks like a function URI. For example lookup would return:

Process Line Functions

The last step is to process all jobs in the queue.

const { onTaskDispatched } = require("firebase-functions/v2/tasks");
exports.processScheduledPosts = onTaskDispatched(
  {
    retryConfig: {
      maxAttempts: 1
    },
    rateLimits: {
      maxConcurrentDispatches: 25
    },
    timeoutSeconds: 480,
    memory: "2GiB"
  },
  async (req) => {
    const { data } = req;
    return processPost(data);
);

We use Firebase’s built-in functionality onTaskDispatched processing each job. You receive the data sent to the function in the request object req and remove i data thing. Processing can be performed on a data set, e.g processPost.

I onTaskDispatched the function requires several parameters:

• retryConfig – How many retries are made if an error occurs.
• rateLimit – Number of concurrent processes allowed.
• timeoutSeconds – The number of seconds before the task ends.
• memory – How much memory is allocated to the task. Please see the pricing for Firebase cloud service.

Firebase Logging in Log Explorer

Another amazing benefit of using the Cloud Task queue is the splitting of logs for each task.

In Google Cloud Log Explorer find the entries made by the task:

Click on the four blue lines:

and select “Show entries in this track”. Now you’ll only see the logs for that specific cloud activity…a huge advantage over trying to sort through mixed logs. Be sure to exclude from the search field any entries other than “trace=…”.

When Should You Use (And Not Use) Firebase Login Functions?

If you need to process tasks independently, asynchronously, and at the same time, Firebase and Cloud Task queues are a good match. Even if you have simple queuing needs, you can use Cloud Tasks. It’s really easy once you get everything set up.

The caveat emptor is the cost. Because you are calling a Firebase function for every queue job you will be paying for every job. You will need to weigh the cost versus value of these Cloud Services, and you may find a simple Firestore line, as mentioned in the use case above, meets your needs.

Overall, Ayrshare’s switch to Firebase Queues with Cloud Tasks was a great new architecture, which increased our stability, speed, and maintenance.

About Ayrshare

Ayrshare is a social media API that allows you to publish posts, get analytics, manage comments, and send direct messages to social media directly from your platform. Read more in our social media API documentation.