package pluginhost import ( "context" "encoding/json" "fmt" "strings" "github.com/router-for-me/CLIProxyAPI/v7/internal/logging" "github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginabi" "github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginapi" log "github.com/sirupsen/logrus" ) type rpcHostHTTPRequest struct { HTTPClientID string `json:"http_client_id,omitempty"` HostCallbackID string `json:"host_callback_id,omitempty"` Method string `json:"method,omitempty"` URL string `json:"url,omitempty"` Headers httpHeader `json:"headers,omitempty"` Body []byte `json:"body,omitempty"` Request *httpRequest `json:"request,omitempty"` } type httpHeader map[string][]string type httpRequest struct { Method string `json:"method,omitempty"` URL string `json:"url,omitempty"` Headers httpHeader `json:"headers,omitempty"` Body []byte `json:"body,omitempty"` } type rpcHostHTTPStreamResponse struct { StatusCode int `json:"status_code"` Headers httpHeader `json:"headers,omitempty"` StreamID string `json:"stream_id,omitempty"` Chunks []pluginapi.HTTPStreamChunk `json:"chunks,omitempty"` } type rpcHostHTTPStreamReadRequest struct { StreamID string `json:"stream_id"` } type rpcHostHTTPStreamReadResponse struct { Payload []byte `json:"payload,omitempty"` Error string `json:"error,omitempty"` Done bool `json:"done,omitempty"` } type rpcHostHTTPStreamCloseRequest struct { StreamID string `json:"stream_id"` } type rpcHostLogRequest struct { HostCallbackID string `json:"host_callback_id,omitempty"` Level string `json:"level,omitempty"` Message string `json:"message,omitempty"` Fields map[string]any `json:"fields,omitempty"` } func (h *Host) callFromPlugin(ctx context.Context, method string, request []byte) ([]byte, error) { switch method { case pluginabi.MethodHostHTTPDo: return h.callHostHTTPDo(ctx, request) case pluginabi.MethodHostHTTPDoStream: return h.callHostHTTPDoStream(ctx, request) case pluginabi.MethodHostHTTPStreamRead: return h.callHostHTTPStreamRead(ctx, request) case pluginabi.MethodHostHTTPStreamClose: return h.callHostHTTPStreamClose(request) case pluginabi.MethodHostStreamEmit: return h.callHostStreamEmit(ctx, request) case pluginabi.MethodHostStreamClose: return h.callHostStreamClose(request) case pluginabi.MethodHostLog: return h.callHostLog(ctx, request) default: return nil, fmt.Errorf("unsupported host callback %s", method) } } func (h *Host) callHostHTTPDo(ctx context.Context, request []byte) ([]byte, error) { httpReq, callbackID, errDecode := decodeHostHTTPRequestWithCallbackID(request) if errDecode != nil { return nil, errDecode } ctx = h.resolveCallbackContext(callbackID, ctx) resp, errDo := h.newHTTPClient(nil).Do(ctx, httpReq) if errDo != nil { return nil, errDo } return marshalRPCResult(resp) } func (h *Host) callHostHTTPDoStream(ctx context.Context, request []byte) ([]byte, error) { httpReq, callbackID, errDecode := decodeHostHTTPRequestWithCallbackID(request) if errDecode != nil { return nil, errDecode } ctx = h.resolveCallbackContext(callbackID, ctx) if ctx == nil { ctx = context.Background() } streamCtx, cancel := context.WithCancel(ctx) resp, errDo := h.newHTTPClient(nil).DoStream(streamCtx, httpReq) if errDo != nil { cancel() return nil, errDo } streamID := "" if h != nil && h.httpStreams != nil { streamID = h.httpStreams.open(resp.Chunks, cancel) } if streamID == "" { cancel() return nil, fmt.Errorf("host http stream bridge is unavailable") } return marshalRPCResult(rpcHostHTTPStreamResponse{ StatusCode: resp.StatusCode, Headers: httpHeader(resp.Headers), StreamID: streamID, }) } func (h *Host) callHostHTTPStreamRead(ctx context.Context, request []byte) ([]byte, error) { var req rpcHostHTTPStreamReadRequest if errUnmarshal := json.Unmarshal(request, &req); errUnmarshal != nil { return nil, fmt.Errorf("decode host http stream read request: %w", errUnmarshal) } if h == nil || h.httpStreams == nil { return nil, fmt.Errorf("host http stream bridge is unavailable") } chunk, done, errRead := h.httpStreams.read(ctx, req.StreamID) if errRead != nil { return nil, errRead } resp := rpcHostHTTPStreamReadResponse{ Payload: append([]byte(nil), chunk.Payload...), Done: done, } if chunk.Err != nil { resp.Error = chunk.Err.Error() resp.Done = true } return marshalRPCResult(resp) } func (h *Host) callHostHTTPStreamClose(request []byte) ([]byte, error) { var req rpcHostHTTPStreamCloseRequest if errUnmarshal := json.Unmarshal(request, &req); errUnmarshal != nil { return nil, fmt.Errorf("decode host http stream close request: %w", errUnmarshal) } if h != nil && h.httpStreams != nil { h.httpStreams.close(req.StreamID) } return marshalRPCResult(rpcEmptyResponse{}) } func decodeHostHTTPRequest(raw []byte) (pluginapi.HTTPRequest, error) { httpReq, _, errDecode := decodeHostHTTPRequestWithCallbackID(raw) return httpReq, errDecode } func decodeHostHTTPRequestWithCallbackID(raw []byte) (pluginapi.HTTPRequest, string, error) { var req rpcHostHTTPRequest if errUnmarshal := json.Unmarshal(raw, &req); errUnmarshal != nil { return pluginapi.HTTPRequest{}, "", fmt.Errorf("decode host http request: %w", errUnmarshal) } if req.Request != nil { return pluginapi.HTTPRequest{ Method: req.Request.Method, URL: req.Request.URL, Headers: map[string][]string(req.Request.Headers), Body: append([]byte(nil), req.Request.Body...), }, req.HostCallbackID, nil } return pluginapi.HTTPRequest{ Method: req.Method, URL: req.URL, Headers: map[string][]string(req.Headers), Body: append([]byte(nil), req.Body...), }, req.HostCallbackID, nil } func (h *Host) callHostStreamEmit(ctx context.Context, request []byte) ([]byte, error) { var req rpcStreamEmitRequest if errUnmarshal := json.Unmarshal(request, &req); errUnmarshal != nil { return nil, fmt.Errorf("decode stream emit request: %w", errUnmarshal) } chunk := pluginapi.ExecutorStreamChunk{Payload: append([]byte(nil), req.Payload...)} if req.Error != "" { chunk.Err = fmt.Errorf("%s", req.Error) } if errEmit := h.streams.emit(ctx, req.StreamID, chunk); errEmit != nil { return nil, errEmit } return marshalRPCResult(rpcEmptyResponse{}) } func (h *Host) callHostStreamClose(request []byte) ([]byte, error) { var req rpcStreamCloseRequest if errUnmarshal := json.Unmarshal(request, &req); errUnmarshal != nil { return nil, fmt.Errorf("decode stream close request: %w", errUnmarshal) } h.streams.close(req.StreamID, req.Error) return marshalRPCResult(rpcEmptyResponse{}) } func (h *Host) callHostLog(ctx context.Context, request []byte) ([]byte, error) { var req rpcHostLogRequest if errUnmarshal := json.Unmarshal(request, &req); errUnmarshal != nil { return nil, fmt.Errorf("decode host log request: %w", errUnmarshal) } ctx = h.resolveCallbackContext(req.HostCallbackID, ctx) message := strings.TrimSpace(req.Message) if message == "" { message = "plugin log" } fields := log.Fields{} for key, value := range req.Fields { key = strings.TrimSpace(key) if key != "" { fields[key] = value } } if requestID := logging.GetRequestID(ctx); requestID != "" { fields["request_id"] = requestID } entry := log.WithFields(fields) switch strings.ToLower(strings.TrimSpace(req.Level)) { case "trace": entry.Trace(message) case "info": entry.Info(message) case "warn", "warning": entry.Warn(message) case "error": entry.Error(message) default: entry.Debug(message) } return marshalRPCResult(rpcEmptyResponse{}) }