// r8r: rust port of n8n.io
// Initial author: Michael Cochran <mcochran@cochranblock.org>
// Built for: Tanner Woodrum, intended cofounder
// License: Unlicense (public domain dedication, see UNLICENSE)
//
// `redis` — KV operations against a Redis server.
// Param names: operation, key, value, ttl, expire, propertyName.
// Credential `redis`: { host, port, password?, user?, database, ssl?,
// disableTlsVerification? }.

use async_trait::async_trait;
use crate::core::error::NodeError;
use crate::core::types::*;
use crate::engine::{ExecuteCtx, Node, NodeFactory};
use redis::AsyncCommands;
use serde_json::{json, Value};
use std::sync::Arc;

pub struct RedisNode;

static META: NodeMetadata = NodeMetadata {
    name: "redis",
    display_name: "Redis",
    version: &[1],
    group: &["database"],
    description: "Get, set, delete keys against a Redis server",
    inputs: 1,
    outputs: 1,
    properties: &[
        PropertyDef { name: "operation", display_name: "Operation", kind: PropertyKind::Options, required: true, default: Some("get") },
        PropertyDef { name: "key", display_name: "Key", kind: PropertyKind::String, required: false, default: None },
        PropertyDef { name: "value", display_name: "Value", kind: PropertyKind::String, required: false, default: None },
        PropertyDef { name: "expire", display_name: "Expire", kind: PropertyKind::Boolean, required: false, default: Some("false") },
        PropertyDef { name: "ttl", display_name: "TTL (seconds)", kind: PropertyKind::Number, required: false, default: Some("60") },
        PropertyDef { name: "propertyName", display_name: "Property Name", kind: PropertyKind::String, required: false, default: Some("propertyName") },
    ],
    credentials: &[CredentialRef { name: "redis", required: true }],
};

#[async_trait]
impl Node for RedisNode {
    fn metadata(&self) -> &'static NodeMetadata { &META }

    async fn execute(&self, ctx: &mut ExecuteCtx<'_>) -> Result<NodeOutput, NodeError> {
        let host = ctx.credential_field("host")?
            .as_str().ok_or_else(|| NodeError::MissingCredential("redis.host".into()))?
            .to_string();
        let port = ctx.credential_field("port").ok()
            .and_then(|v| v.as_u64()).unwrap_or(6379);
        let user = ctx.credential_field("user").ok()
            .and_then(|v| v.as_str().map(|s| s.to_string())).unwrap_or_default();
        let password = ctx.credential_field("password").ok()
            .and_then(|v| v.as_str().map(|s| s.to_string()));
        let database = ctx.credential_field("database").ok()
            .and_then(|v| v.as_u64()).unwrap_or(0);
        let ssl = ctx.credential_field("ssl").ok()
            .and_then(|v| v.as_bool()).unwrap_or(false);
        let scheme = if ssl { "rediss" } else { "redis" };
        let auth = match (user.is_empty(), password) {
            (false, Some(p)) => format!("{}:{}@", user, p),
            (true, Some(p)) => format!(":{}@", p),
            _ => String::new(),
        };
        let url = format!("{}://{}{}:{}/{}", scheme, auth, host, port, database);

        let client = redis::Client::open(url.as_str())
            .map_err(|e| NodeError::Other(format!("redis: client {}: {}", host, e)))?;
        let mut conn = client.get_multiplexed_async_connection().await
            .map_err(|e| NodeError::Other(format!("redis: connect {}: {}", host, e)))?;

        let operation = ctx.param_opt_str("operation").unwrap_or_else(|| "get".into());
        let item_json: Value = match operation.as_str() {
            "get" => {
                let key = ctx.param_str("key")?;
                let prop = ctx.param_opt_str("propertyName").unwrap_or_else(|| "propertyName".into());
                let val: Option<String> = conn.get(&key).await
                    .map_err(|e| NodeError::Other(format!("redis get: {}", e)))?;
                json!({ prop: val })
            }
            "set" => {
                let key = ctx.param_str("key")?;
                let value = ctx.param_str("value")?;
                let expire = ctx.node_def.parameters.get("expire")
                    .and_then(|v| v.as_bool()).unwrap_or(false);
                if expire {
                    let ttl = ctx.node_def.parameters.get("ttl")
                        .and_then(|v| v.as_u64()).unwrap_or(60);
                    let _: () = conn.set_ex(&key, &value, ttl).await
                        .map_err(|e| NodeError::Other(format!("redis set_ex: {}", e)))?;
                } else {
                    let _: () = conn.set(&key, &value).await
                        .map_err(|e| NodeError::Other(format!("redis set: {}", e)))?;
                }
                json!({ "key": key, "value": value, "ok": true })
            }
            "delete" => {
                let key = ctx.param_str("key")?;
                let n: u64 = conn.del(&key).await
                    .map_err(|e| NodeError::Other(format!("redis del: {}", e)))?;
                json!({ "key": key, "deleted": n })
            }
            "incr" => {
                let key = ctx.param_str("key")?;
                let n: i64 = conn.incr(&key, 1).await
                    .map_err(|e| NodeError::Other(format!("redis incr: {}", e)))?;
                json!({ "key": key, "value": n })
            }
            "keys" => {
                let pattern = ctx.param_opt_str("key").unwrap_or_else(|| "*".into());
                let keys: Vec<String> = conn.keys(&pattern).await
                    .map_err(|e| NodeError::Other(format!("redis keys: {}", e)))?;
                json!({ "keys": keys })
            }
            "publish" => {
                let key = ctx.param_str("key")?;
                let value = ctx.param_str("value")?;
                let n: u64 = conn.publish(&key, &value).await
                    .map_err(|e| NodeError::Other(format!("redis publish: {}", e)))?;
                json!({ "channel": key, "subscribers": n })
            }
            other => return Err(NodeError::Other(format!("redis: unknown operation `{}`", other))),
        };

        Ok(vec![vec![NodeItem { json: item_json, ..Default::default() }]])
    }
}

inventory::submit! { NodeFactory::new("redis", || Arc::new(RedisNode)) }
