// ✅ Declare a priority queue — x-max-priority must be set at declaration time
await channel.assertQueue('notifications', {
  durable: true,
  arguments: {
    'x-max-priority': 5,            // 5 priority levels: 0 (lowest) to 5 (highest)
    'x-queue-type':   'classic',    // Priority queues are classic only — NOT quorum
  }
})

// Publish with explicit priority
channel.sendToQueue('notifications', Buffer.from(JSON.stringify({
  type:   'password-reset',
  userId: 'u-42',
  token:  'abc123',
})), {
  priority:   5,      // highest — delivered before any lower-priority messages
  persistent: true,
})

channel.sendToQueue('notifications', Buffer.from(JSON.stringify({
  type:   'weekly-digest',
  userId: 'u-42',
})), {
  priority:   2,      // delivered after priority-5 messages are drained
  persistent: true,
})

channel.sendToQueue('notifications', Buffer.from(JSON.stringify({
  type:   'marketing-campaign',
  campaignId: 'CAMP-2026-Q4',
})), {
  priority:   0,      // lowest — processed only when higher-priority queue is empty
  persistent: true,
})

// ✅ Better pattern for most use cases: separate physical queues per priority tier
// Worker processes high-priority queue first, falls back to low when empty
async function smartConsumer(): Promise<void> {
  // Check high-priority first
  const highMsg = await channel.get('notifications.high', { noAck: false })
  if (highMsg) {
    await processNotification(JSON.parse(highMsg.content.toString()))
    channel.ack(highMsg)
    return
  }
  // Fall back to low-priority
  const lowMsg = await channel.get('notifications.low', { noAck: false })
  if (lowMsg) {
    await processNotification(JSON.parse(lowMsg.content.toString()))
    channel.ack(lowMsg)
  }
}

// ✅ Kafka: separate topics + weighted polling
const highConsumer = kafka.consumer({ groupId: 'notifications-high' })
const lowConsumer  = kafka.consumer({ groupId: 'notifications-low'  })

await highConsumer.subscribe({ topic: 'notifications.high' })
await lowConsumer.subscribe({  topic: 'notifications.low'  })

// Poll high-priority 4× for every 1 low-priority poll
let pollCount = 0
setInterval(async () => {
  pollCount++
  if (pollCount % 5 !== 0) {
    // 4 out of 5 cycles: drain high-priority
    await highConsumer.run({ eachMessage: processNotification })
  } else {
    // 1 out of 5 cycles: process low-priority
    await lowConsumer.run({ eachMessage: processNotification })
  }
}, 100)

// ✅ Delayed delivery via TTL + DLX — no plugin required
async function scheduleNotification(
  payload:   object,
  delayMs:   number
): Promise<void> {
  const holdingQueue = `notifications.delay.${delayMs}`

  // Declare a holding queue with this exact delay duration
  await channel.assertQueue(holdingQueue, {
    durable: true,
    arguments: {
      'x-message-ttl':             delayMs,
      'x-dead-letter-exchange':    '',                     // default exchange
      'x-dead-letter-routing-key': 'notifications.ready', // routes here after TTL
      'x-expires':                 delayMs + 60_000,       // auto-delete queue after delay + buffer
    }
  })

  channel.sendToQueue(holdingQueue, Buffer.from(JSON.stringify(payload)), {
    persistent: true,
  })
  // Message sits in holdingQueue for delayMs, then routes to notifications.ready
}

await scheduleNotification({ type: 'reminder', userId: 'u-42' }, 60_000)   // 1 minute
await scheduleNotification({ type: 'reminder', userId: 'u-43' }, 300_000)  // 5 minutes

// ✅ SQS per-message delay — up to 15 minutes (900 seconds)
import { SQSClient, SendMessageCommand } from '@aws-sdk/client-sqs'

const sqs = new SQSClient({ region: 'us-east-1' })

async function scheduleRenewalReminder(
  userId:    string,
  delayDays: number
): Promise<void> {
  const delaySeconds = Math.min(delayDays * 86400, 900) // SQS max: 900s = 15 min

  if (delayDays * 86400 > 900) {
    // For delays > 15 minutes: store in scheduler DB, publish to SQS at correct time
    await schedulerDb.insert({
      userId,
      scheduledAt: new Date(Date.now() + delayDays * 86400 * 1000),
      payload:     { type: 'renewal-reminder', userId },
    })
    return
  }

  await sqs.send(new SendMessageCommand({
    QueueUrl:     process.env.NOTIFICATIONS_QUEUE_URL!,
    MessageBody:  JSON.stringify({ type: 'renewal-reminder', userId }),
    DelaySeconds: delaySeconds,
  }))
}

// ❌ Kafka timestamp-hold — blocks the partition thread, starves other messages
await consumer.run({
  eachMessage: async ({ message, pause }) => {
    const scheduledAt = Number(message.headers?.['scheduled-at'])
    const now = Date.now()

    if (scheduledAt > now) {
      const waitMs = scheduledAt - now
      if (waitMs > 30_000) {
        // More than 30 seconds: pause partition and reschedule check
        const resume = pause()
        setTimeout(resume, Math.min(waitMs, 30_000))
        return
      }
      // Wait inline — blocks entire partition thread
      await new Promise(resolve => setTimeout(resolve, waitMs))
    }
    await processScheduledJob(JSON.parse(message.value!.toString()))
  }
})

// ✅ BullMQ: durable job scheduling with priority and delay
import { Queue, Worker } from 'bullmq'
import Redis from 'ioredis'

const connection = new Redis({ host: 'redis', port: 6379, maxRetriesPerRequest: null })

const notificationQueue = new Queue('notifications', { connection })

// Priority job (lower number = higher priority in BullMQ)
await notificationQueue.add('password-reset', {
  userId: 'u-42',
  token:  'abc123',
}, {
  priority: 1,     // highest priority
  attempts: 3,
  backoff:  { type: 'exponential', delay: 1000 },
})

// Delayed job: send in 7 days
await notificationQueue.add('renewal-reminder', {
  userId:         'u-43',
  subscriptionId: 'sub-789',
}, {
  delay:    7 * 24 * 3600 * 1000,   // 7 days in ms — stored durably in Redis
  priority: 10,                      // low priority
  attempts: 5,
})

// Repeatable (cron) job
await notificationQueue.add('weekly-digest', {}, {
  repeat: { pattern: '0 8 * * 1' },  // every Monday at 08:00
})

// Worker: processes jobs from the queue
const worker = new Worker('notifications', async (job) => {
  switch (job.name) {
    case 'password-reset':
      await emailService.sendPasswordReset(job.data)
      break
    case 'renewal-reminder':
      await emailService.sendRenewalReminder(job.data)
      break
    case 'weekly-digest':
      await emailService.sendWeeklyDigest()
      break
  }
}, {
  connection,
  concurrency: 10,
})

worker.on('failed', (job, err) => {
  console.error(`Job ${job?.id} failed: ${err.message}`)
})