Class TencentCmqQueueProvider

src/cloud/queue.ts:56–136

}

export class TencentCmqQueueProvider implements QueueProvider {
  readonly providerName = 'tdmq-cmq' as const;
  private readonly receipts = new Map<string, string>();

  constructor(
    private readonly store: JobStore = jobStore,
    private readonly client: TencentCmqClient = createTencentCmqClient()
  ) {}

  async publish(message: QueueJobMessage, options: PublishOptions = {}): Promise<void> {
    await this.client.sendMessage(message, options.delaySeconds);
  }

  async claimNext(input: ClaimJobInput): Promise<CloudJob | undefined> {
    const received = await this.client.receiveMessage();
    if (!received) return undefined;

    const existing = await this.store.get(received.body.jobId);
    if (!existing) {
      await this.client.deleteMessage(received.receiptHandle);
      return undefined;
    }
    const job = await this.store.claim(received.body.jobId, input);
    if (!job) {
      const latest = (await this.store.get(received.body.jobId)) || existing;
      if (isTerminalJobStatus(latest.status)) {
        await this.client.deleteMessage(received.receiptHandle);
      } else if (latest.status === 'processing') {
        await this.deferWatchdogMessage(received.receiptHandle, latest);
      } else if (latest.status === 'retry_wait') {
        await this.deferWatchdogMessage(received.receiptHandle, latest);
      } else {
        await this.client.deleteMessage(received.receiptHandle);
      }
      return undefined;
    }
    this.receipts.set(job.id, received.receiptHandle);
    return job;
  }

  async complete(job: CloudJob): Promise<void> {
    await this.deleteReceipt(job.id);
  }

  async release(job: CloudJob, options: PublishOptions = {}): Promise<void> {
    await this.deleteReceipt(job.id);
    if (job.status === 'retry_wait') {
      await this.publish(
        {
          jobId: job.id,
          tenantId: job.tenantId,
          taskType: job.taskType,
          attempt: job.attempts,
          traceId: job.externalJobId,
        },
        options
      );
    }