// 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)
//
// `mongodb` — find / insert / update / delete / aggregate against a
// MongoDB collection. Param names lifted from upstream MongoDb.node.ts:
// operation, collection, query, fields, options, sort, limit, skip,
// updateKey, useDotNotation, dateFields.
// Credential `mongoDb`: { configurationType, connectionString, host,
// database, user, password, port, tls, ca }.

use async_trait::async_trait;
use crate::core::error::NodeError;
use crate::core::types::*;
use crate::engine::{ExecuteCtx, Node, NodeFactory};
use mongodb::bson::{doc, Document};
use mongodb::Client;
use serde_json::Value;
use std::sync::Arc;

pub struct MongoDbNode;

static META: NodeMetadata = NodeMetadata {
    name: "mongodb",
    display_name: "MongoDB",
    version: &[1],
    group: &["input"],
    description: "Find, insert and update documents in MongoDB",
    inputs: 1,
    outputs: 1,
    properties: &[
        PropertyDef { name: "operation", display_name: "Operation", kind: PropertyKind::Options, required: true, default: Some("find") },
        PropertyDef { name: "collection", display_name: "Collection", kind: PropertyKind::String, required: true, default: None },
        PropertyDef { name: "query", display_name: "Query", kind: PropertyKind::Json, required: false, default: None },
        PropertyDef { name: "fields", display_name: "Fields", kind: PropertyKind::String, required: false, default: None },
        PropertyDef { name: "options", display_name: "Options", kind: PropertyKind::Json, required: false, default: None },
        PropertyDef { name: "sort", display_name: "Sort", kind: PropertyKind::Json, required: false, default: None },
        PropertyDef { name: "limit", display_name: "Limit", kind: PropertyKind::Number, required: false, default: Some("100") },
        PropertyDef { name: "skip", display_name: "Skip", kind: PropertyKind::Number, required: false, default: Some("0") },
        PropertyDef { name: "updateKey", display_name: "Update Key", kind: PropertyKind::String, required: false, default: Some("_id") },
    ],
    credentials: &[CredentialRef { name: "mongoDb", required: true }],
};

fn json_to_doc(v: &Value) -> Document {
    match mongodb::bson::to_document(v) {
        Ok(d) => d,
        Err(_) => Document::new(),
    }
}

fn build_uri(ctx: &mut ExecuteCtx<'_>) -> Result<(String, String), NodeError> {
    let cfg = ctx.credential_field("configurationType").ok()
        .and_then(|v| v.as_str().map(|s| s.to_string()))
        .unwrap_or_else(|| "values".into());
    let database = ctx.credential_field("database")?
        .as_str().ok_or_else(|| NodeError::MissingCredential("mongoDb.database".into()))?
        .to_string();
    if cfg == "connectionString" {
        let cs = ctx.credential_field("connectionString")?
            .as_str().ok_or_else(|| NodeError::MissingCredential("mongoDb.connectionString".into()))?
            .to_string();
        return Ok((cs, database));
    }
    let host = ctx.credential_field("host").ok()
        .and_then(|v| v.as_str().map(|s| s.to_string()))
        .unwrap_or_else(|| "127.0.0.1".into());
    let port = ctx.credential_field("port").ok()
        .and_then(|v| v.as_u64()).unwrap_or(27017);
    let user = ctx.credential_field("user").ok()
        .and_then(|v| v.as_str().map(|s| s.to_string()));
    let pass = ctx.credential_field("password").ok()
        .and_then(|v| v.as_str().map(|s| s.to_string()));
    let auth = match (user, pass) {
        (Some(u), Some(p)) if !u.is_empty() => format!("{}:{}@", u, p),
        _ => "".to_string(),
    };
    let tls = ctx.credential_field("tls").ok()
        .and_then(|v| v.as_bool()).unwrap_or(false);
    let suffix = if tls { "?tls=true" } else { "" };
    Ok((format!("mongodb://{}{}:{}/{}{}", auth, host, port, database, suffix), database))
}

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

    async fn execute(&self, ctx: &mut ExecuteCtx<'_>) -> Result<NodeOutput, NodeError> {
        let (uri, database) = build_uri(ctx)?;
        let collection_name = ctx.param_str("collection")?;
        let operation = ctx.param_opt_str("operation").unwrap_or_else(|| "find".into());
        let query_v = ctx.node_def.parameters.get("query").cloned().unwrap_or(Value::Null);

        let client = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            Client::with_uri_str(&uri),
        ).await
            .map_err(|_| NodeError::Other("mongodb connect: timed out".into()))?
            .map_err(|e| NodeError::Other(format!("mongodb connect: {}", e)))?;
        let coll = client.database(&database).collection::<Document>(&collection_name);

        let json_value = match operation.as_str() {
            "find" => {
                let filter = if query_v.is_null() { Document::new() } else { json_to_doc(&query_v) };
                let mut cur = coll.find(filter).await
                    .map_err(|e| NodeError::Other(format!("mongodb find: {}", e)))?;
                let mut out = Vec::new();
                use futures_util::StreamExt;
                while let Some(d) = cur.next().await {
                    if let Ok(d) = d {
                        out.push(serde_json::to_value(d).unwrap_or(Value::Null));
                    }
                }
                Value::Array(out)
            }
            "insert" => {
                let doc = if query_v.is_null() { Document::new() } else { json_to_doc(&query_v) };
                let res = coll.insert_one(doc).await
                    .map_err(|e| NodeError::Other(format!("mongodb insert: {}", e)))?;
                serde_json::json!({ "insertedId": res.inserted_id.to_string() })
            }
            "delete" => {
                let filter = if query_v.is_null() { Document::new() } else { json_to_doc(&query_v) };
                let res = coll.delete_many(filter).await
                    .map_err(|e| NodeError::Other(format!("mongodb delete: {}", e)))?;
                serde_json::json!({ "deletedCount": res.deleted_count })
            }
            "update" => {
                let filter = if query_v.is_null() { Document::new() } else { json_to_doc(&query_v) };
                let updates = ctx.node_def.parameters.get("options").cloned()
                    .unwrap_or(Value::Object(Default::default()));
                let update_doc = doc! { "$set": json_to_doc(&updates) };
                let res = coll.update_many(filter, update_doc).await
                    .map_err(|e| NodeError::Other(format!("mongodb update: {}", e)))?;
                serde_json::json!({
                    "matchedCount": res.matched_count,
                    "modifiedCount": res.modified_count,
                })
            }
            other => return Err(NodeError::Other(format!("mongodb: {} not implemented", other))),
        };
        Ok(vec![vec![NodeItem { json: json_value, ..Default::default() }]])
    }
}

inventory::submit! { NodeFactory::new("mongodb", || Arc::new(MongoDbNode)) }
