#!/usr/bin/env babel-node --harmony

// import firequeue from 'firequeue'

import firequeue from './src'
import { concat, throughSync } from 'stream-util'
import parallel from 'concurrent-transform'

const queue = firequeue.init('https://firequeue-test.firebaseio.com')
const logger = (fn) => throughSync(function(data) {
  console.log(fn(data))
  this.push(data)
})

// create some jobs
// in a real application you might want to use push to generate random job id instead
queue.incoming.child('job1').set({ task: 'task1', data: { name: 'job1' } })
queue.incoming.child('job2').set({ task: 'task1', data: { name: 'job2' } })
queue.incoming.child('job3').set({ task: 'task1', data: { name: 'job3' }, delayed: '20s' })
queue.incoming.child('job4').set({ task: 'task2', data: { name: 'job4' } })
queue.incoming.child('job5').set({ task: 'task3', data: { name: 'job5' } })

// listen to job updates
queue.jobs
  .child('job1/state')
  .on('value', (s) => console.log(`job1 changed state to ${s.val()}`))

// log 'queued'
// log 'activated'
// log 'completed'

// start queue engine
queue
  .start()
  .pipe(logger(({ task, key, state }) => `task: ${task}, job: ${key}, state: ${state}`))

// log task: task1, job: job1, state: queued
// log task: task1, job: job2, state: queued
// log task: task1, job: job3, state: delayed
// ...

// process task1
const task1 = queue
  .read('task1')
  .pipe(queue.process((job) => {
    // do some work with job.key(), job.val()
    return Promise.resolve()
  }))
  .pipe(logger(({ task, key, state }) => `task: ${task}, job: ${key}, state: ${state}`))

// log task: task1, job: job1, state: completed
// log task: task1, job: job2, state: completed
// ...

// process task2 with maxAttempts and backoff
const task2 = queue
  .read('task2')
  .pipe(queue.maxAttempts(2))
  .pipe(queue.backoff('2s'))  // wait 2s before retrying
  .pipe(queue.process((job) => {
    console.log('do some work with', job.key(), job.val())
    const attempts = job.child('attempts').val() || 0
    return  attempts < 2
      ? Promise.reject()
      : Promise.resolve()
  }))

// process task3 with a concurrency of 10
const task3 = queue
  .read('task3')
  .pipe(parallel(queue.process((job) => {
    console.log('do some work with', job.key(), job.val())
    return Promise.resolve()
  }), 10))


// remove completed jobs
concat(task1, task2, task3)
// log task: task1, job: job1, state: cleaned
// log task: task1, job: job2, state: cleaned
// ...

// 30sec later...
setTimeout(() => {
  queue.stop().then(() => {
    console.log('queue was stopped successfuly')
  })
}, 30000)
  .pipe(queue.clean('completed'))
  .pipe(logger(({ task, key, state }) => `task: ${task}, job: ${key}, state: ${state}`))
