/**
 * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
 * <p>
 * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.gnu.org/licenses/lgpl-3.0.txt
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.easyagents.flow.core.node;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.easyagents.flow.core.chain.Chain;
import com.easyagents.flow.core.chain.DataType;
import com.easyagents.flow.core.chain.Parameter;
import com.easyagents.flow.core.filestoreage.FileStorage;
import com.easyagents.flow.core.filestoreage.FileStorageManager;
import com.easyagents.flow.core.util.IoBulkhead;
import com.easyagents.flow.core.util.OkHttpClientUtil;
import com.easyagents.flow.core.util.StringUtil;
import com.easyagents.flow.core.util.TextTemplate;
import okhttp3.*;

import java.io.ByteArrayOutputStream;
import java.io.FilterInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class HttpNode extends BaseNode {
    private static final long serialVersionUID = 1L;


    private static final long DEFAULT_MAX_TEXT_RESPONSE_BYTES = 64L * 1024L * 1024L;
    private static final long DEFAULT_MAX_FILE_RESPONSE_BYTES = 2L * 1024L * 1024L * 1024L;
    private static final String MAX_RESPONSE_BYTES_PROPERTY = "tinyflow.http.max-response-bytes";
    private static final String MAX_FILE_RESPONSE_BYTES_PROPERTY = "tinyflow.http.max-file-response-bytes";

    private String url;
    private String method;

    private List<Parameter> headers;

    private String bodyType;
    private List<Parameter> formData;
    private List<Parameter> formUrlencoded;
    private String bodyJson;
    private String rawBody;

    public static String mapToQueryString(Map<String, Object> map) {
        if (map == null || map.isEmpty()) {
            return "";
        }

        StringBuilder stringBuilder = new StringBuilder();

        for (String key : map.keySet()) {
            if (StringUtil.noText(key)) {
                continue;
            }
            if (stringBuilder.length() > 0) {
                stringBuilder.append("&");
            }
            stringBuilder.append(key.trim());
            stringBuilder.append("=");
            Object value = map.get(key);
            stringBuilder.append(value == null ? "" : urlEncode(value.toString().trim()));
        }
        return stringBuilder.toString();
    }

    public static String urlEncode(String string) {
        try {
            return URLEncoder.encode(string, "UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException(e);
        }
    }

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public String getMethod() {
        return method;
    }

    public void setMethod(String method) {
        this.method = method;
    }

    public List<Parameter> getHeaders() {
        return headers;
    }

    public void setHeaders(List<Parameter> headers) {
        this.headers = headers;
    }

    public String getBodyType() {
        return bodyType;
    }

    public void setBodyType(String bodyType) {
        this.bodyType = bodyType;
    }

    public List<Parameter> getFormData() {
        return formData;
    }

    public void setFormData(List<Parameter> formData) {
        this.formData = formData;
    }

    public List<Parameter> getFormUrlencoded() {
        return formUrlencoded;
    }

    public void setFormUrlencoded(List<Parameter> formUrlencoded) {
        this.formUrlencoded = formUrlencoded;
    }

    public String getBodyJson() {
        return bodyJson;
    }

    public void setBodyJson(String bodyJson) {
        this.bodyJson = bodyJson;
    }

    public String getRawBody() {
        return rawBody;
    }

    public void setRawBody(String rawBody) {
        this.rawBody = rawBody;
    }

    @Override
    public Map<String, Object> execute(Chain chain) {
        int maxRetry = supportsAutomaticRetry(method) ? 5 : 1;
        long retryInterval = 2000L;

        int attempt = 0;
        Throwable lastError = null;

        while (attempt < maxRetry) {
            attempt++;

            try {
                return doExecute(chain);
            } catch (Throwable ex) {

                lastError = ex;

                // 判断是否需要重试
                if (attempt >= maxRetry || !shouldRetry(ex)) {
                    throw wrapAsRuntime(ex, attempt);
                }

                try {
                    long waitMs = Math.min(
                            retryInterval * (1L << (attempt - 1)),
                            10_000L // 最大 10 秒
                    );
                    Thread.sleep(waitMs);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException("HTTP retry interrupted", ie);
                }
            }
        }

        // 理论上不会走到这里
        throw wrapAsRuntime(lastError, attempt);
    }


    protected boolean shouldRetry(Throwable ex) {
        if (ex instanceof HttpServerErrorException) {
            int code = ((HttpServerErrorException) ex).getStatusCode();
            return code == 503 || code == 504; // 只对特定 5xx 重试
        }

        // 1. IO 异常（超时、连接失败、Socket 问题）
        if (ex instanceof IOException) {
            return true;
        }

        // 2. 包装过的异常
        Throwable cause = ex.getCause();
        return cause instanceof IOException;
    }

    /**
     * 判断当前方法是否允许节点内部自动重试。
     *
     * @param requestMethod HTTP 方法
     * @return 仅无业务副作用的读取类方法返回 {@code true}
     */
    protected boolean supportsAutomaticRetry(String requestMethod) {
        return StringUtil.noText(requestMethod)
                || "GET".equalsIgnoreCase(requestMethod)
                || "HEAD".equalsIgnoreCase(requestMethod)
                || "OPTIONS".equalsIgnoreCase(requestMethod);
    }

    private RuntimeException wrapAsRuntime(Throwable ex, int attempt) {
        if (ex instanceof RuntimeException) {
            return (RuntimeException) ex;
        }
        return new RuntimeException(
                String.format("HttpNode[%s] failed after %d attempt(s)", getName(), attempt),
                ex
        );
    }


    public Map<String, Object> doExecute(Chain chain) throws IOException {

        Map<String, Object> argsMap =
                chain.getExecutionState().resolveParameters(this);
        String newUrl = TextTemplate.of(url)
                .formatToString(
                        chain.getExecutionState()
                                .buildTemplateRootMaps(argsMap));

        Request.Builder reqBuilder = new Request.Builder().url(newUrl);

        Map<String, Object> headersMap =
                chain.getExecutionState().resolveParameters(
                        this, headers, argsMap);
        headersMap.forEach((s, o) -> reqBuilder.addHeader(s, String.valueOf(o)));

        if (StringUtil.noText(method) || "GET".equalsIgnoreCase(method)) {
            reqBuilder.method("GET", null);
        } else {
            reqBuilder.method(method.toUpperCase(), getRequestBody(chain, argsMap));
        }

        // 节点层统一计算总尝试次数，禁用客户端隐式重试，避免乘法式放大。
        OkHttpClient okHttpClient =
                OkHttpClientUtil.buildNoRetryClient();
        // 共享客户端拦截器持有请求许可直到响应体关闭，避免同一请求重复领取许可。
        try (Response response = okHttpClient.newCall(
                     reqBuilder.build()).execute()) {

            // 服务器异常
            if (response.code() >= 500 && response.code() < 600) {
                throw new HttpServerErrorException(response.code(), response.message());
            }

            Map<String, Object> result = new HashMap<>();
            result.put("statusCode", response.code());

            Map<String, String> responseHeaders = new HashMap<>();
            Headers headers = response.headers();
            for (String name : headers.names()) {
                responseHeaders.put(name, response.header(name));
            }
            result.put("headers", responseHeaders);

            ResponseBody body = response.body();
            if (body == null) {
                result.put("body", null);
                return result;
            }

            DataType bodyDataType = null;
            List<Parameter> outputDefs = getOutputDefs();
            if (outputDefs != null) {
                for (Parameter outputDef : outputDefs) {
                    if ("body".equalsIgnoreCase(outputDef.getName())) {
                        bodyDataType = outputDef.getDataType();
                        break;
                    }
                }
            }

            if (bodyDataType == null) {
                try (IoBulkhead.Permit ignored =
                             IoBulkhead.responseAggregation().acquire(
                                     IoBulkhead.targetForUrl(newUrl))) {
                    result.put("body", readTextBody(
                            body, resolveMaxTextResponseBytes()));
                }
            } else if (bodyDataType == DataType.Object || bodyDataType.getValue().startsWith("Array")) {
                try (IoBulkhead.Permit ignored =
                             IoBulkhead.responseAggregation().acquire(
                                     IoBulkhead.targetForUrl(newUrl))) {
                    result.put("body", JSON.parse(readTextBody(
                            body, resolveMaxTextResponseBytes())));
                }
            } else if (bodyDataType == DataType.File) {
                long maxFileResponseBytes = resolveMaxFileResponseBytes();
                validateDeclaredResponseSize(body, maxFileResponseBytes);
                try (InputStream stream = limitResponseStream(body.byteStream(), maxFileResponseBytes)) {
                    FileStorage fileStorage = FileStorageManager.getInstance().getFileStorage();
                    try (IoBulkhead.Permit ignored =
                                 IoBulkhead.storage().acquire(
                                         "storage:http-response")) {
                        String fileUrl = fileStorage.saveFile(
                                stream, responseHeaders, this, chain);
                        result.put("body", fileUrl);
                    }
                }
            } else {
                try (IoBulkhead.Permit ignored =
                             IoBulkhead.responseAggregation().acquire(
                                     IoBulkhead.targetForUrl(newUrl))) {
                    result.put("body", readTextBody(
                            body, resolveMaxTextResponseBytes()));
                }
            }
            return result;
        }
    }

    private RequestBody getRequestBody(Chain chain, Map<String, Object> formatArgs) {
        if ("json".equals(bodyType)) {
            String bodyJsonString = TextTemplate.of(bodyJson)
                    .formatToString(
                            chain.getExecutionState()
                                    .buildTemplateRootMaps(formatArgs),
                            true);
            JSONObject jsonObject = JSON.parseObject(bodyJsonString);
            return RequestBody.create(jsonObject.toString(), MediaType.parse("application/json"));
        }

        if ("x-www-form-urlencoded".equals(bodyType)) {
            Map<String, Object> formUrlencodedMap =
                    chain.getExecutionState().resolveParameters(
                            this, formUrlencoded);
            String bodyString = mapToQueryString(formUrlencodedMap);
            return RequestBody.create(bodyString, MediaType.parse("application/x-www-form-urlencoded"));
        }

        if ("form-data".equals(bodyType)) {
            Map<String, Object> formDataMap =
                    chain.getExecutionState().resolveParameters(
                            this, formData, formatArgs);

            MultipartBody.Builder builder = new MultipartBody.Builder()
                    .setType(MultipartBody.FORM);

            formDataMap.forEach((s, o) -> {
//                if (o instanceof File) {
//                    File f = (File) o;
//                    RequestBody body = RequestBody.create(f, MediaType.parse("application/octet-stream"));
//                    builder.addFormDataPart(s, f.getName(), body);
//                } else if (o instanceof InputStream) {
//                    RequestBody body = new HttpClient.InputStreamRequestBody(MediaType.parse("application/octet-stream"), (InputStream) o);
//                    builder.addFormDataPart(s, s, body);
//                } else if (o instanceof byte[]) {
//                    builder.addFormDataPart(s, s, RequestBody.create((byte[]) o));
//                } else {
//                    builder.addFormDataPart(s, String.valueOf(o));
//                }
                builder.addFormDataPart(s, String.valueOf(o));
            });

            return builder.build();
        }

        if ("raw".equals(bodyType)) {
            String rawBodyString = TextTemplate.of(rawBody)
                    .formatToString(
                            chain.getExecutionState()
                                    .buildTemplateRootMaps(formatArgs));
            return RequestBody.create(rawBodyString, null);
        }
        //none
        return RequestBody.create("", null);
    }

    /**
     * 在宽松响应大小保护下读取文本响应。
     *
     * @param body HTTP 响应体
     * @return 响应文本
     * @throws IOException 响应读取失败或超出限制时抛出
     */
    protected String readTextBody(ResponseBody body, long maxResponseBytes) throws IOException {
        validateDeclaredResponseSize(body, maxResponseBytes);
        Charset charset = body.contentType() == null
                ? StandardCharsets.UTF_8
                : body.contentType().charset(StandardCharsets.UTF_8);
        int initialCapacity = body.contentLength() > 0L
                ? (int) Math.min(body.contentLength(), 64L * 1024L)
                : 8 * 1024;
        try (InputStream input = limitResponseStream(body.byteStream(), maxResponseBytes);
             ByteArrayOutputStream output = new ByteArrayOutputStream(initialCapacity)) {
            byte[] buffer = new byte[64 * 1024];
            int read;
            while ((read = input.read(buffer)) != -1) {
                output.write(buffer, 0, read);
            }
            return output.toString(charset);
        }
    }

    /**
     * 校验响应声明长度，避免继续处理已知超限内容。
     *
     * @param body HTTP 响应体
     * @param maxResponseBytes 最大允许字节数
     * @throws IOException 声明长度超过限制时抛出
     */
    private void validateDeclaredResponseSize(ResponseBody body, long maxResponseBytes) throws IOException {
        long contentLength = body.contentLength();
        if (maxResponseBytes > 0L && contentLength > maxResponseBytes) {
            throw responseSizeExceeded(maxResponseBytes);
        }
    }

    /**
     * 为响应流增加按实际读取字节数执行的限制。
     *
     * @param inputStream 原始响应流
     * @param maxResponseBytes 最大允许字节数
     * @return 受限响应流
     */
    private InputStream limitResponseStream(InputStream inputStream, long maxResponseBytes) {
        if (maxResponseBytes <= 0L) {
            return inputStream;
        }
        return new LimitedResponseInputStream(inputStream, maxResponseBytes);
    }

    /**
     * 获取文本或 JSON 响应字节数上限。
     *
     * @return 最大允许字节数
     */
    private long resolveMaxTextResponseBytes() {
        return Long.getLong(MAX_RESPONSE_BYTES_PROPERTY, DEFAULT_MAX_TEXT_RESPONSE_BYTES);
    }

    /**
     * 获取文件响应字节数上限。
     *
     * @return 最大允许字节数
     */
    private long resolveMaxFileResponseBytes() {
        return Long.getLong(MAX_FILE_RESPONSE_BYTES_PROPERTY, DEFAULT_MAX_FILE_RESPONSE_BYTES);
    }

    /**
     * 创建统一的响应超限异常。
     *
     * @param maxResponseBytes 最大允许字节数
     * @return 响应超限异常
     */
    private IOException responseSizeExceeded(long maxResponseBytes) {
        return new IOException("HTTP response body exceeds limit: " + maxResponseBytes + " bytes");
    }

    public static class HttpServerErrorException extends IOException {
        private final int statusCode;

        public HttpServerErrorException(int statusCode, String message) {
            super("HTTP " + statusCode + ": " + message);
            this.statusCode = statusCode;
        }

        public int getStatusCode() {
            return statusCode;
        }
    }

    /**
     * 按实际读取量限制 HTTP 响应大小的输入流。
     */
    private final class LimitedResponseInputStream extends FilterInputStream {

        private final long maxBytes;
        private long consumed;

        /**
         * 创建受限响应流。
         *
         * @param inputStream 原始输入流
         * @param maxBytes 最大允许字节数
         */
        private LimitedResponseInputStream(InputStream inputStream, long maxBytes) {
            super(inputStream);
            this.maxBytes = maxBytes;
        }

        /**
         * 读取单个字节并校验累计读取量。
         *
         * @return 读取的字节或 {@code -1}
         * @throws IOException 读取失败或超过限制时抛出
         */
        @Override
        public int read() throws IOException {
            int value = super.read();
            if (value >= 0) {
                recordRead(1L);
            }
            return value;
        }

        /**
         * 批量读取并校验累计读取量。
         *
         * @param buffer 目标缓冲区
         * @param offset 写入偏移
         * @param length 最大读取长度
         * @return 实际读取长度或 {@code -1}
         * @throws IOException 读取失败或超过限制时抛出
         */
        @Override
        public int read(byte[] buffer, int offset, int length) throws IOException {
            long remainingWithProbe = maxBytes == Long.MAX_VALUE
                    ? Long.MAX_VALUE
                    : maxBytes - consumed + 1L;
            int allowed = (int) Math.min(
                    Math.max(0L, remainingWithProbe),
                    (long) length);
            if (allowed <= 0) {
                throw responseSizeExceeded(maxBytes);
            }
            int count = super.read(buffer, offset, allowed);
            if (count > 0) {
                recordRead(count);
            }
            return count;
        }

        /**
         * 记录实际读取量。
         *
         * @param count 本次读取字节数
         * @throws IOException 超过限制时抛出
         */
        private void recordRead(long count) throws IOException {
            consumed += count;
            if (consumed > maxBytes) {
                throw responseSizeExceeded(maxBytes);
            }
        }
    }


    @Override
    public String toString() {
        return "HttpNode{" +
                "url='" + url + '\'' +
                ", method='" + method + '\'' +
                ", headers=" + headers +
                ", bodyType='" + bodyType + '\'' +
                ", formData=" + formData +
                ", formUrlencoded=" + formUrlencoded +
                ", bodyJson='" + bodyJson + '\'' +
                ", rawBody='" + rawBody + '\'' +
                ", parameters=" + parameters +
                ", outputDefs=" + outputDefs +
                ", id='" + id + '\'' +
                ", name='" + name + '\'' +
                ", description='" + description + '\'' +
                ", condition=" + condition +
                ", validator=" + validator +
                ", loopEnable=" + loopEnable +
                ", loopIntervalMs=" + loopIntervalMs +
                ", loopBreakCondition=" + loopBreakCondition +
                ", maxLoopCount=" + maxLoopCount +
                ", retryEnable=" + retryEnable +
                ", resetRetryCountAfterNormal=" + resetRetryCountAfterNormal +
                ", maxRetryCount=" + maxRetryCount +
                ", retryIntervalMs=" + retryIntervalMs +
                ", computeCostExpr='" + computeCostExpr + '\'' +
                '}';
    }
}
