import { BaseCommand, flags } from '@adonisjs/core/ace'
import { CommandOptions } from '@adonisjs/core/types/ace'
import indexerConfig from '#config/indexer'
import QueueService from '#services/QueueService'
import RedisService from '#services/RedisService'
import DeferredService, { isDeferred } from '#services/DeferredService'

const BATCH = 2000

/** Rebuilds the wait list (KEYS[1]) without the ids in the set KEYS[2], in one atomic pass. */
const FILTER_LIST = `
local ids = redis.call('LRANGE', KEYS[1], 0, -1)
local keep = {}
for _, id in ipairs(ids) do
  if redis.call('SISMEMBER', KEYS[2], id) == 0 then keep[#keep + 1] = id end
end
redis.call('DEL', KEYS[1])
for i = 1, #keep, 5000 do
  redis.call('RPUSH', KEYS[1], unpack(keep, i, math.min(i + 4999, #keep)))
end
return #ids - #keep`

/**
 * Removes waiting body jobs whose message-id is on the skip or defer list,
 * e.g. after adding a pattern (CONCEPT.md D24, D28). Deferred ones are
 * recorded in deferred_files first, so they can still be processed later.
 */
export default class BodyQueuePrune extends BaseCommand {
  public static commandName = 'body-queue:prune'
  public static description = 'Drops waiting body jobs that are now skipped or deferred.'

  public static options: CommandOptions = {
    startApp: true,
  }

  @flags.boolean({ description: 'Only count' })
  declare dryRun: boolean

  @flags.boolean({ description: 'Drop deferred jobs without recording them (already recorded by an earlier run)' })
  declare skipRecord: boolean

  public async run() {
    const redis = RedisService.client
    const queue = QueueService.bodyQueue
    const waitKey = queue.toKey('wait')
    let skipped = 0
    let deferred = 0
    let scanned = 0
    const dropped: string[] = []
    try {
      const ids: string[] = await redis.lrange(waitKey, 0, -1)
      for (let i = 0; i < ids.length; i += BATCH) {
        const batch = ids.slice(i, i + BATCH)
        const data: [Error | null, string | null][] = await batch
          .reduce((pipeline, id) => pipeline.hget(queue.toKey(id), 'data'), redis.pipeline())
          .exec()
        const drop: string[] = []
        const toDefer = new Map<string, { articleNumber: number; header: Record<string, string> }[]>()
        batch.forEach((id, j) => {
          const job = data[j][1] ? JSON.parse(data[j][1]!) : null
          if (!job?.header) return
          const messageId = job.header['message-id'] ?? ''
          if (isDeferred(job.header)) {
            const list = toDefer.get(job.group) ?? []
            list.push({ articleNumber: Number(job.articleNumber), header: job.header })
            toDefer.set(job.group, list)
            drop.push(id)
            deferred++
          } else if (indexerConfig.skipBodyMessageIds.some((re) => re.test(messageId))) {
            drop.push(id)
            skipped++
          }
        })
        scanned += batch.length
        if (this.dryRun || drop.length === 0) continue
        if (!this.skipRecord) for (const [group, articles] of toDefer) await DeferredService.record(group, articles)
        dropped.push(...drop)
      }
      if (dropped.length > 0) {
        const dropKey = queue.toKey('prune-ids')
        await redis.del(dropKey)
        for (let i = 0; i < dropped.length; i += BATCH) await redis.sadd(dropKey, ...dropped.slice(i, i + BATCH))
        await redis.eval(FILTER_LIST, 2, waitKey, dropKey)
        await redis.del(dropKey)
        for (let i = 0; i < dropped.length; i += BATCH) {
          await redis.unlink(...dropped.slice(i, i + BATCH).map((id) => queue.toKey(id)))
        }
      }
      this.logger.info(
        `Scanned ${scanned} waiting body jobs: ${this.dryRun ? 'would drop' : 'dropped'} ${deferred} deferred and ${skipped} skipped.`
      )
    } finally {
      await QueueService.closeAll()
      await RedisService.close()
    }
  }
}