Method dispatchToConsumer

packages/cli/src/util/dev/queue-broker.ts:400–475

(
    message: StoredMessage,
    group: ConsumerGroup
  )

  }

  private async dispatchToConsumer(
    message: StoredMessage,
    group: ConsumerGroup
  ): Promise<void> {
    const groupDeliveries = this.deliveryState.get(group.id);
    if (!groupDeliveries) return;

    const state = groupDeliveries.get(message.messageId);
    if (!state || state.status === 'acked') return;

    if (state.deliveryCount >= group.maxDeliveries) {
      output.debug(
        `queues: message ${message.messageId} exceeded maxDeliveries (${group.maxDeliveries}) for group "${group.name}", dropping`
      );
      groupDeliveries.delete(message.messageId);
      this.maybeCleanupMessage(message.messageId);
      return;
    }

    const upstream = group.serviceOriginFn();
    if (!upstream) {
      // Service not ready yet, retry later
      state.visibleAt = Date.now() + group.retryAfterMs;
      return;
    }

    const receiptHandle = randomBytes(16).toString('hex');
    state.status = 'in-flight';
    state.receiptHandle = receiptHandle;
    state.deliveryCount++;
    state.leaseExpiresAt = Date.now() + DEFAULT_VISIBILITY_TIMEOUT;

    const now = new Date().toISOString();
    const expiresAt = new Date(
      new Date(message.createdAt).getTime() + message.retentionMs
    ).toISOString();

    output.debug(
      `queues: dispatching v2beta callback to worker "${group.name}" at ${upstream}`
    );

    try {
      const response = await directFetch(`${upstream}/`, {
        method: 'POST',
        headers: {
          'content-type': message.contentType,
          'ce-type': 'com.vercel.queue.v2beta',
          'ce-specversion': '1.0',
          'ce-source': `/topic/${message.queueName}/consumer/${group.name}`,
          'ce-id': message.messageId,
          'ce-time': now,
          'ce-vqsmessageid': message.messageId,
          'ce-vqsqueuename': message.queueName,
          'ce-vqsconsumergroup': group.name,
          'ce-vqsreceipthandle': receiptHandle,
          'ce-vqsdeliverycount': String(state.deliveryCount),
          'ce-vqscreatedat': message.createdAt,
          'ce-vqsexpiresat': expiresAt,