Method enqueue

packages/cli/src/util/dev/queue-broker.ts:132–221

(
    queueName: string,
    payload: Buffer,
    contentType: string,
    options?: EnqueueOptions
  )

  }

  enqueue(
    queueName: string,
    payload: Buffer,
    contentType: string,
    options?: EnqueueOptions
  ): { messageId: string } {
    const messageId = randomBytes(16).toString('hex');
    const retentionMs =
      (options?.retentionSeconds ?? 0) > 0
        ? options!.retentionSeconds! * 1000
        : DEFAULT_RETENTION;
    const idempotencyRecordKey = options?.idempotencyKey
      ? `${queueName}:${options.idempotencyKey}`
      : undefined;

    if (idempotencyRecordKey) {
      const record = this.idempotencyRecords.get(idempotencyRecordKey);
      if (record && record.expiresAt > Date.now()) {
        this.duplicateMessages.set(messageId, {
          queueName,
          originalMessageId: record.messageId,
          expiresAt: Date.now() + retentionMs,
        });
        output.debug(
          `queues: skipped duplicate message for queue "${queueName}"`
        );
        return { messageId };
      }
      if (record) {
        this.idempotencyRecords.delete(idempotencyRecordKey);
      }
    }

    if (idempotencyRecordKey) {
      this.idempotencyRecords.set(idempotencyRecordKey, {
        messageId,
        expiresAt: Date.now() + retentionMs,
      });
    }

    const message: StoredMessage = {
      messageId,
      payload,
      contentType,
      queueName,
      createdAt: new Date().toISOString(),
      retentionMs,
    };

    this.messages.set(messageId, message);
    output.debug(
      `queues: stored message ${messageId} for queue "${queueName}"`
    );

    const delaySeconds = options?.delaySeconds ?? 0;
    const matchingGroups = this.consumerGroups.filter(g =>
      g.topicRegex.test(queueName)
    );