optimize[syncer]: optimize returning error when it occur (#18613)

### Summary

As title

internal/handler/connector.go

@@ -105,7 +105,9 @@ func connectorErrorResponse(c *gin.Context, err error) bool {
 	case errors.Is(err, service.ErrConnectorNotFound):
 		common.ResponseWithCodeData(c, common.CodeDataError, nil, "Can't find this Connector!")
 	case errors.Is(err, service.ErrConnectorTestUnsupported):
-		common.ResponseWithCodeData(c, common.CodeArgumentError, false, err.Error())
+		common.ResponseWithCodeData(c, common.CodeNotImplemented, false, err.Error())
+	case errors.Is(err, service.ErrConnectorSourceNotImplemented):
+		common.ResponseWithCodeData(c, common.CodeNotImplemented, false, err.Error())
 	default:
 		common.ResponseWithHttpCodeData(c, http.StatusInternalServerError, common.CodeServerError, nil, err.Error())
 	}
@@ -381,7 +383,7 @@ func (h *ConnectorHandler) TestConnector(c *gin.Context) {
 	}
 
 	err := h.connectorService.TestConnector(ctx, connectorID, user.ID, request)
-	if errors.Is(err, service.ErrConnectorTestUnsupported) {
+	if errors.Is(err, service.ErrConnectorTestUnsupported) || errors.Is(err, service.ErrConnectorSourceNotImplemented) {
 		connectorErrorResponse(c, err)
 		return
 	}
@@ -397,7 +399,7 @@ func (h *ConnectorHandler) TestConnector(c *gin.Context) {
 			common.ResponseWithCodeData(c, common.CodeDataError, false, err.Error())
 			return
 		}
-		common.ResponseWithCodeData(c, common.CodeServerError, false, "REST API connector validation failed, please check logs.")
+		common.ResponseWithCodeData(c, common.CodeServerError, false, err.Error())
 		return
 	}
 	if connectorErrorResponse(c, err) {

internal/handler/connector_test.go

@@ -241,9 +241,10 @@ func TestConnectorHandlerTestConnector(t *testing.T) {
 	gin.SetMode(gin.TestMode)
 
 	tests := []struct {
-		name     string
-		err      error
-		wantCode common.ErrorCode
+		name        string
+		err         error
+		wantCode    common.ErrorCode
+		wantMessage string
 	}{
 		{
 			name:     "success",
@@ -263,7 +264,13 @@ func TestConnectorHandlerTestConnector(t *testing.T) {
 		{
 			name:     "unsupported source",
 			err:      service.ErrConnectorTestUnsupported,
-			wantCode: common.CodeArgumentError,
+			wantCode: common.CodeNotImplemented,
+		},
+		{
+			name:        "source not implemented",
+			err:         fmt.Errorf("%w: seafile", service.ErrConnectorSourceNotImplemented),
+			wantCode:    common.CodeNotImplemented,
+			wantMessage: "connector source is not implemented: seafile",
 		},
 		{
 			name:     "schema validation failure",
@@ -281,9 +288,10 @@ func TestConnectorHandlerTestConnector(t *testing.T) {
 			wantCode: common.CodeDataError,
 		},
 		{
-			name:     "unexpected failure",
-			err:      fmt.Errorf("boom"),
-			wantCode: common.CodeServerError,
+			name:        "unexpected failure",
+			err:         fmt.Errorf("boom"),
+			wantCode:    common.CodeServerError,
+			wantMessage: "boom",
 		},
 	}
 
@@ -310,6 +318,9 @@ func TestConnectorHandlerTestConnector(t *testing.T) {
 			if body["code"] != float64(tt.wantCode) {
 				t.Fatalf("code=%v want=%v body=%v", body["code"], tt.wantCode, body)
 			}
+			if tt.wantMessage != "" && body["message"] != tt.wantMessage {
+				t.Fatalf("message=%v want=%v body=%v", body["message"], tt.wantMessage, body)
+			}
 		})
 	}
 }

internal/service/connector.go

@@ -80,6 +80,8 @@ var (
 	ErrConnectorNoAuth = errors.New("no authorization")
 	// ErrConnectorTestUnsupported is returned for connector sources without a settings validator.
 	ErrConnectorTestUnsupported = errors.New("connector test is not supported for this source")
+	// ErrConnectorSourceNotImplemented is returned for connector sources not registered in the Go syncer.
+	ErrConnectorSourceNotImplemented = errors.New("connector source is not implemented")
 )
 
 // ConnectorService connector service
@@ -422,6 +424,10 @@ func (s *ConnectorService) TestConnector(ctx context.Context, connectorID, userI
 	}
 	connector, err := s.connectorRegistry.OpenFromConfig(source, connectorConfig)
 	if err != nil {
+		var unsupported *syncerconnector.UnsupportedSourceError
+		if errors.As(err, &unsupported) {
+			return fmt.Errorf("%w: %s", ErrConnectorSourceNotImplemented, unsupported.Source)
+		}
 		return err
 	}
 	validator, ok := connector.(syncerconnector.SettingValidator)

internal/service/connector_test_connection_test.go

@@ -148,8 +148,8 @@ func TestConnectorServiceTestConnectorRejectsUnsupportedSource(t *testing.T) {
 		"source": "unknown",
 		"config": entity.JSONMap{"ok": true},
 	})
-	if err == nil || !strings.Contains(err.Error(), `unsupported connector source "unknown"`) {
-		t.Fatalf("error = %v, want unsupported source", err)
+	if !errors.Is(err, ErrConnectorSourceNotImplemented) || !strings.Contains(err.Error(), "unknown") {
+		t.Fatalf("error = %v, want source not implemented", err)
 	}
 }
 

internal/syncer/connector/registry.go

@@ -18,11 +18,28 @@ package connector
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"ragflow/internal/dao"
 	"sync"
 )
 
+// ErrUnsupportedSource is returned when no Go connector is registered for a source.
+var ErrUnsupportedSource = errors.New("unsupported connector source")
+
+// UnsupportedSourceError identifies a connector source that is not implemented.
+type UnsupportedSourceError struct {
+	Source string
+}
+
+func (e *UnsupportedSourceError) Error() string {
+	return fmt.Sprintf("%s %q", ErrUnsupportedSource, e.Source)
+}
+
+func (e *UnsupportedSourceError) Unwrap() error {
+	return ErrUnsupportedSource
+}
+
 // Factory creates a connector for a task context.
 type Factory func(ctx context.Context, taskContext dao.SyncTaskContext) (Connector, error)
 
@@ -69,7 +86,7 @@ func (r *Registry) OpenFromConfig(source string, config map[string]any) (Connect
 	factory := r.configFactories[source]
 	r.mu.RUnlock()
 	if factory == nil {
-		return nil, fmt.Errorf("unsupported connector source %q", source)
+		return nil, &UnsupportedSourceError{Source: source}
 	}
 	return factory(config)
 }
@@ -80,7 +97,7 @@ func (r *Registry) openSource(ctx context.Context, source string, taskContext da
 	factory := r.factories[source]
 	r.mu.RUnlock()
 	if factory == nil {
-		return nil, fmt.Errorf("unsupported connector source %q", source)
+		return nil, &UnsupportedSourceError{Source: source}
 	}
 	return factory(ctx, taskContext)
 }

internal/syncer/connector/registry_test.go

@@ -2,6 +2,7 @@ package connector
 
 import (
 	"context"
+	"errors"
 	"ragflow/internal/dao"
 	"ragflow/internal/entity"
 	"strings"
@@ -23,7 +24,7 @@ func TestRegistryOpenFromConfig(t *testing.T) {
 	}
 
 	_, err = registry.OpenFromConfig("missing", map[string]any{})
-	if err == nil || !strings.Contains(err.Error(), `unsupported connector source "missing"`) {
+	if err == nil || !errors.Is(err, ErrUnsupportedSource) || !strings.Contains(err.Error(), `unsupported connector source "missing"`) {
 		t.Fatalf("unsupported source error = %v", err)
 	}
 }