use std::sync::Arc;

use aok::{OK, Void};
use fred::{
  interfaces::ClientLike,
  prelude::FredResult,
  types::{CustomCommand, Key, Value},
};
use xkv::R;

use crate::{Kv, StreamItem, auto_new, parse_stream};

pub struct ReadGroup {
  pub stream: Key,
  pub group: String,
  pub consumer: String,
  pub block_ms: u64,
  pub claim_idle_ms: u64,
  pub count: u64,
  pub max_retry: u64,
}

impl ReadGroup {
  pub fn new(
    stream: impl Into<Key>,
    group: impl Into<String>,
    consumer: impl Into<String>,
    block_sec: u64,
    claim_idle_sec: u64,
    count: u64,
    max_retry: u64,
  ) -> Self {
    Self {
      stream: stream.into(),
      group: group.into(),
      consumer: consumer.into(),
      block_ms: block_sec * 1000,
      claim_idle_ms: claim_idle_sec * 1000,
      count,
      max_retry,
    }
  }

  pub async fn run<
    R1: Future<Output = aok::Void> + Send + 'static,
    R2: Future<Output = aok::Void> + Send + 'static,
  >(
    &self,
    parse: fn(Arc<Kv>) -> R1,
    on_error: fn(Arc<Kv>, String) -> R2,
  ) -> Void {
    let group = &self.group;
    let stream = &self.stream;
    loop {
      let count_str = self.count.to_string();
      let block_str = self.block_ms.to_string();
      let claim_str = self.claim_idle_ms.to_string();

      let args: Vec<Value> = vec![
        "GROUP".into(),
        group.clone().into(),
        self.consumer.clone().into(),
        "COUNT".into(),
        count_str.into(),
        "BLOCK".into(),
        block_str.into(),
        "CLAIM".into(),
        claim_str.into(),
        "STREAMS".into(),
        stream.clone().into(),
        ">".into(),
      ];

      let li: FredResult<Vec<StreamItem>> = R
        .custom(
          CustomCommand::new("XREADGROUP", stream.as_bytes(), true),
          args,
        )
        .await
        .map(parse_stream);

      if let Some(li) = auto_new(stream, group, li).await? {
        if li.is_empty() {
          break;
        }

        let mut ing = Vec::with_capacity(li.len());

        for StreamItem { id, retry, kv, .. } in li {
          let kv = Arc::new(kv);
          ing.push((id, retry, kv.clone(), tokio::spawn((parse)(kv))));
        }

        let mut id_li: Vec<String> = Vec::new();

        for (id, retry, kv, task) in ing {
          let err = match task.await {
            Ok(Err(err)) => err.to_string(),
            Err(err) => err.to_string(),
            _ => {
              id_li.push(id);
              continue;
            }
          };
          log::error!("{id} retry {retry} {err}");
          if retry > self.max_retry {
            id_li.push(id);
            tokio::spawn(async move { xerr::log!(on_error(kv, err).await) });
          }
        }
        if !id_li.is_empty() {
          crate::rm_id_li(&self.stream, &self.group, id_li).await?;
        }
      }
    }
    OK
  }
}
