use std::time::{Duration, Instant};

use anyhow::Result;
use futures_util::StreamExt;
use rivet_util::Id;

/// Time to delay a workflow from retrying after an error
pub const RETRY_TIMEOUT_MS: usize = 1000;
pub const WORKFLOW_TIMEOUT: Duration = Duration::from_secs(60);

use crate::{
	ctx::OperationCtx,
	db::{BumpSubSubject, DatabaseHandle, WorkflowData},
	error::WorkflowError,
	operation::{Operation, OperationInput},
	utils::tags::AsTags,
	workflow::Workflow,
};

/// Polls the database for the workflow.
/// 60 second timeout.
pub async fn wait_for_workflow_output<W: Workflow>(
	db: &DatabaseHandle,
	workflow_id: Id,
) -> Result<W::Output> {
	tracing::debug!(?workflow_id, "waiting for workflow");

	let mut bump_sub = db
		.bump_sub(BumpSubSubject::WorkflowComplete { workflow_id })
		.await?;
	let mut interval = tokio::time::interval(db.sub_workflow_poll_interval());

	// Skip first tick, we wait after the db call instead of before
	interval.tick().await;

	tokio::time::timeout(WORKFLOW_TIMEOUT, async {
		loop {
			// Check if state finished
			let workflow = db
				.get_workflows(vec![workflow_id])
				.await?
				.into_iter()
				.next()
				.ok_or(WorkflowError::WorkflowNotFound)?;
			if let Some(output) = workflow.parse_output::<W>()? {
				return Ok(output);
			}

			// Poll and wait for a wake at the same time
			tokio::select! {
				_ = bump_sub.next() => {},
				_ = interval.tick() => {},
			}
		}
	})
	.await?
}

/// Finds the first incomplete workflow with the given tags.
pub async fn find_workflow<W: Workflow>(
	db: &DatabaseHandle,
	tags: impl AsTags,
) -> Result<Option<Id>> {
	db.find_workflow(W::NAME, &tags.as_tags()?)
		.await
		.map_err(Into::into)
}

/// Finds the first incomplete workflow for each (name, tags) pair in a single batch transaction.
pub async fn find_workflows(
	db: &DatabaseHandle,
	queries: &[(&str, serde_json::Value)],
) -> Result<Vec<Option<Id>>> {
	db.find_workflows(queries).await.map_err(Into::into)
}

/// Finds the first incomplete workflow with the given tags.
pub async fn get_workflows(
	db: &DatabaseHandle,
	workflow_ids: Vec<Id>,
) -> Result<Vec<WorkflowData>> {
	db.get_workflows(workflow_ids).await.map_err(Into::into)
}

pub async fn op<I>(
	db: &DatabaseHandle,
	config: &rivet_config::Config,
	pools: &rivet_pools::Pools,
	cache: &rivet_cache::Cache,
	ray_id: Id,
	from_workflow: bool,
	input: I,
) -> Result<<<I as OperationInput>::Operation as Operation>::Output>
where
	I: OperationInput,
	<I as OperationInput>::Operation: Operation<Input = I>,
{
	tracing::debug!(?input, "operation call");

	// Record metrics
	crate::metrics::OPERATION_PENDING
		.with_label_values(&[I::Operation::NAME])
		.inc();
	crate::metrics::OPERATION_TOTAL
		.with_label_values(&[I::Operation::NAME])
		.inc();

	let start_instant = Instant::now();

	let ctx = OperationCtx::new(
		db.clone(),
		config,
		pools,
		cache,
		ray_id,
		from_workflow,
		I::Operation::NAME,
	)?;

	let res = tokio::time::timeout(I::Operation::TIMEOUT, I::Operation::run(&ctx, &input))
		.await
		.map_err(|_| WorkflowError::OperationTimeout(I::Operation::NAME, 0))
		.map(|res| res.map_err(|err| WorkflowError::OperationFailure(I::Operation::NAME, err)));

	// Record metrics
	{
		let error_str = match &res {
			Ok(Err(_)) => {
				// TODO: Temporarily don't record err to reduce metrics cardinality
				// let error_str = err.to_string();

				// crate::metrics::OPERATION_ERRORS
				// 	.with_label_values(&[I::Operation::NAME, error_str.as_str()])
				// 	.inc();

				// error_str
				String::new()
			}
			Ok(_) => String::new(),
			Err(_) => "timeout".to_string(),
		};

		// Other request metrics
		let dt = start_instant.elapsed().as_secs_f64();
		crate::metrics::OPERATION_PENDING
			.with_label_values(&[I::Operation::NAME])
			.dec();
		crate::metrics::OPERATION_DURATION
			.with_label_values(&[I::Operation::NAME, error_str.as_str()])
			.observe(dt);
	}

	let res = res?;

	tracing::debug!(?res, "operation response");

	res.map_err(Into::into)
}
