meteor-job-collection v2.0

Persistent Reactive Job Queue for Meteor 3.x

✨ What's New in v2.0

🚨 Upgrading from v1.x?

// OLD (v1.x with Fibers)
const job = myJobs.getJob(id);
              job.done();

// NEW (v2.x)
const job = await myJobs.getJob(id);
await job.done();

📦 Installation

For Meteor Apps (Recommended)

meteor add strigops:job-collection

For Standalone Node.js Workers

npm install @strigo/meteor-job-collection

🚀 Quick Start

Server Setup

import { JobCollection } from 'meteor/strigops:job-collection';

// Create job collection
const myJobs = new JobCollection('myJobQueue');

// Set up permissions
myJobs.setJobAllow({
  admin: (userId) => !!userId  // Authenticated users only
});

// Start server on startup
Meteor.startup(async () => {
  await myJobs.startJobServer();
});

// Publish jobs (optional)
Meteor.publish('allJobs', function() {
  return myJobs.find({});
});

Create and Submit Jobs

import { Job } from 'meteor/strigops:job-collection';

// Create a job with async/await
async function scheduleEmail() {
  const job = new Job(myJobs, 'sendEmail', {
    to: '[email protected]',
    subject: 'Hello',
    body: 'Welcome!'
  });

  const id = await job
    .priority('normal')
    .retry({ retries: 5, wait: 15*60*1000 })  // 15 min between retries
    .save();

  return id;
}

Process Jobs (Workers)

// Worker with async/await
myJobs.processJobs(
  'sendEmail',
  { concurrency: 4 },
  async (job, callback) => {
    try {
      await sendEmail(job.data);
      await job.done();
    } catch (error) {
      await job.fail(error.message);
    }
    callback();  // Always call callback!
  }
);

📖 Core Concepts

Job Lifecycle

waiting → ready → running → completed
   ↓        ↓        ↓
paused   ready    failed → waiting (retry)
   ↓                 ↓
waiting          cancelled

Job Configuration

const job = new Job(myJobs, 'processImage', data)
  .priority('high')              // low, normal, medium, high, critical
  .retry({
    retries: 5,
    wait: 5*60*1000,             // 5 minutes between retries
    backoff: 'exponential'        // or 'constant'
  })
  .repeat({
    repeats: 10,                  // or Job.forever
    wait: 60*60*1000             // 1 hour between repeats
  })
  .delay(30*1000)                // Delay 30 seconds
  .depends([job1, job2])         // Wait for dependencies
  .after(new Date('2024-01-01')) // Run after date
  .save();

🔧 TypeScript Support

import { Job, JobCollection } from 'meteor/strigops:job-collection';

// Define job data types
interface EmailJobData {
  to: string;
  subject: string;
  body: string;
}

// Type-safe job creation
const job = new Job<EmailJobData>(myJobs, 'sendEmail', {
  to: '[email protected]',
  subject: 'Hello',
  body: 'Welcome!'
});

// Type-safe workers
myJobs.processJobs<EmailJobData>(
  'sendEmail',
  async (job, callback) => {
    const { to, subject, body } = job.data;  // Fully typed!
    await sendEmail(to, subject, body);
    await job.done();
    callback();
  }
);

🎯 API Reference

Job Methods

// With async/await
const id = await job.save();
await job.done(result);
await job.fail(error);
await job.log('message', { level: 'info' });
await job.progress(50, 100);
await job.pause();
await job.resume();
await job.cancel();
await job.restart();
await job.remove();

// With callbacks (backward compatible)
job.save((err, id) => { /* ... */ });
job.done(result, (err, success) => { /* ... */ });

JobCollection Methods

// Get jobs
await myJobs.getJob(id)
await myJobs.getJobs([id1, id2, id3])
await myJobs.getWork(type, options)

// Bulk operations
await myJobs.readyJobs([id1, id2])
await myJobs.pauseJobs([id1, id2])
await myJobs.resumeJobs([id1, id2])
await myJobs.cancelJobs([id1, id2])
await myJobs.restartJobs([id1, id2])
await myJobs.removeJobs([id1, id2])

// Server control
await myJobs.startJobServer()
await myJobs.shutdownJobServer({ timeout: 60000 })

// Worker queue
const workers = myJobs.processJobs(type, options, worker)

JobQueue Methods (Workers)

const workers = myJobs.processJobs('type', options, worker);

workers.pause()          // Pause processing
workers.resume()         // Resume processing
workers.trigger()        // Manually trigger work check
workers.length()         // Jobs waiting in queue
workers.running()        // Jobs currently running
workers.idle()           // True if no jobs
workers.full()           // True if at max concurrency

// Graceful shutdown
workers.shutdown({ level: 'soft' }, () => {
  console.log('Shutdown complete');
});

🔒 Security

// Use setJobAllow/setJobDeny (or allow/deny for backward compatibility)
myJobs.setJobAllow({
  // Admin: full control
  admin: (userId, method, params) => {
    return Roles.userIsInRole(userId, 'admin');
  },
  
  // Manager: can manage existing jobs
  manager: (userId, method, params) => {
    return Roles.userIsInRole(userId, 'manager');
  },
  
  // Creator: can create new jobs
  creator: (userId, method, params) => {
    return userId !== null;
  },
  
  // Worker: can get work and update status
  worker: (userId, method, params) => {
    return Roles.userIsInRole(userId, 'worker');
  }
});

// Deny rules (override allow)
myJobs.setJobDeny({
  admin: (userId) => userId === 'bannedUserId'
});

// Note: You can also use allow/deny for backward compatibility
// myJobs.allow({ ... }) and myJobs.deny({ ... }) work the same way

📡 Remote Workers (Node.js)

// worker.js
import DDP from 'ddp';
import DDPLogin from 'ddp-login';
import { Job } from '@strigo/meteor-job-collection';

const ddp = new DDP({
  host: 'localhost',
  port: 3000,
  use_ejson: true
});

Job.setDDP(ddp);

ddp.connect(async (err) => {
  if (err) throw err;
  
  await DDPLogin(ddp, { /* credentials */ });
  
  const workers = Job.processJobs(
    'myQueue',
    'sendEmail',
    { concurrency: 10 },
    async (job, callback) => {
      await sendEmail(job.data);
      await job.done();
      callback();
    }
  );
});

🔄 Advanced Features

Job Dependencies

// Job 2 waits for Job 1
const job1 = await new Job(myJobs, 'download', data).save();
const job2 = await new Job(myJobs, 'process', data)
  .depends([job1])
  .save();

Scheduled Jobs (Later.js)

const job = new Job(myJobs, 'dailyReport', {})
  .repeat({
    schedule: myJobs.later.parse.text('at 9:00 am every weekday')
  })
  .save();

Progress Tracking

myJobs.processJobs('process', async (job, callback) => {
  const total = 100;
  for (let i = 0; i < total; i++) {
    await processItem(i);
    await job.progress(i + 1, total);
  }
  await job.done();
  callback();
});

Logging

// Set up file logging
import fs from 'fs';
const logStream = fs.createWriteStream('jobs.log', { flags: 'a' });
myJobs.setLogStream(logStream);

// Or use event-based logging
myJobs.events.on('call', (msg) => {
  console.log(`${msg.method}:`, msg.returnVal);
});

🧹 Maintenance

Cleaning Up Old Jobs

// Periodic cleanup
async function cleanupJobs() {
  const weekAgo = new Date(Date.now() - 7*24*60*60*1000);
  const oldJobs = await myJobs
    .find({
      status: { $in: ['completed', 'cancelled'] },
      updated: { $lt: weekAgo }
    })
    .fetchAsync();
  
  const ids = oldJobs.map(j => j._id);
  await myJobs.removeJobs(ids);
}

// Schedule cleanup job
new Job(myJobs, 'cleanup', {})
  .repeat({
    schedule: myJobs.later.parse.text('at 3:00 am')
  })
  .save();

myJobs.processJobs('cleanup', async (job, callback) => {
  await cleanupJobs();
  await job.done();
  callback();
});

Monitoring

// Count jobs by status
const stats = await Promise.all([
  myJobs.find({ status: 'waiting' }).countAsync(),
  myJobs.find({ status: 'running' }).countAsync(),
  myJobs.find({ status: 'completed' }).countAsync(),
  myJobs.find({ status: 'failed' }).countAsync()
]);

console.log('Jobs:', {
  waiting: stats[0],
  running: stats[1],
  completed: stats[2],
  failed: stats[3]
});

⚡ Performance Tips

   myJobs.processJobs('type', { concurrency: 10 }, worker);

   await myJobs.pauseJobs(manyIds);  // Better than loop

   await myJobs.createIndexAsync({ type: 1, created: -1 });

   myJobs.processJobs('type', { prefetch: 5 }, worker);

🐛 Troubleshooting

Jobs Not Processing?

// Check job status
const ready = await myJobs.find({ status: 'ready' }).countAsync();
console.log('Ready jobs:', ready);

// Check worker status
console.log('Running:', workers.running());
console.log('Queued:', workers.length());

// Monitor errors
myJobs.events.on('error', (msg) => {
  console.error('Error:', msg.error);
});

Jobs Stuck in Running?

Performance Issues?

🏗️ Building from Source

# Clone repository
git clone https://github.com/strigo/meteor-job-collection.git
cd meteor-job-collection

# Install dependencies
npm install

# Build TypeScript
npm run build

# Test
meteor test-packages ./

📚 Documentation

Job Document Schema

{
  _id: JobId,
  runId: JobId | null,
  type: string,
  status: 'waiting' | 'paused' | 'ready' | 'running' | 'failed' | 'cancelled' | 'completed',
  data: object,
  result?: object,
  failures?: object[],
  priority: number,
  depends: JobId[],
  resolved: JobId[],
  after: Date,
  updated: Date,
  created: Date,
  workTimeout?: number,
  expiresAfter?: Date,
  log?: LogEntry[],
  progress: { completed: number, total: number, percent: number },
  retries: number,
  retried: number,
  retryUntil: Date,
  retryWait: number,
  retryBackoff: 'constant' | 'exponential',
  repeats: number,
  repeated: number,
  repeatUntil: Date,
  repeatWait: number | LaterJSSchedule
}

DDP Methods

📄 License

🤝 Contributing

Development Setup

git clone YOUR_REPO_URL
cd meteor-job-collection
npm install
npm run watch  # Auto-compile on changes

Running Tests

meteor test-packages ./

🔗 Links