Queues

Message queues for coordinating scraping tasks. Push work items to a queue, pull them from workers, and acknowledge completion.

Creating a Queue

import { ScrayleClient } from '@scrayle/sdk';

const client = new ScrayleClient();

const queue = await client.storage.queue.create({
  name: 'url-queue',
  description: 'URLs to scrape',
  runId: 'run-001',
  actorId: 'my-actor',
});

console.log(queue.id);

Pushing Messages

const { msgId } = await client.storage.queue.push(queue.id, {
  name: 'scrape-page',
  payload: JSON.stringify({
    url: 'https://example.com/page-1',
    format: 'markdown',
  }),
  retry: 3,
  timeout: 30000,
  deadline: Date.now() + 3600000, // 1 hour deadline
});

console.log('Message ID:', msgId);

Pulling Messages

const messages = await client.storage.queue.pull(queue.id, 5); // Pull up to 5

for (const msg of messages) {
  const data = JSON.parse(msg.payload);
  console.log('Processing:', data.url);

  // Do work...

  // Acknowledge completion
  await client.storage.queue.ack(queue.id, msg.id);
}

Queue Statistics

const q = await client.storage.queue.get('url-queue', queue.id);
console.log('Pending:', q.stats.pending);
console.log('Running:', q.stats.running);
console.log('Success:', q.stats.success);
console.log('Failed:', q.stats.failed);

Managing Queues

// List all queues
const queues = await client.storage.queue.list({
  page: 1, pageSize: 20,
});

// Update queue name/description
await client.storage.queue.update(queue.id, {
  name: 'renamed-queue',
  description: 'Updated description',
});

// Delete a queue
await client.storage.queue.delete(queue.id);

Message Lifecycle

1. Push - Message enters the queue (pending)
2. Pull - Worker picks up the message (running)
3. Ack - Worker acknowledges completion (success)
4. Timeout - If not acked within timeout, message re-enters queue
5. Retry - Failed messages are retried up to the configured limit
6. Deadline - Messages that exceed deadline are marked failed

API Reference