ragflow/internal/service/connector.go

//
//  Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
//
//  Licensed under the Apache License, Version 2.0 (the "License");
//  you may not use this file except in compliance with the License.
//  You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
//  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 service

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"html"
	"io"
	"net/http"
	"net/url"
	"ragflow/internal/engine/kvrocks"
	syncerconnector "ragflow/internal/syncer/connector"
	"strings"
	"time"

	"go.uber.org/zap"
	"gorm.io/gorm"

	"ragflow/internal/common"
	"ragflow/internal/dao"
	"ragflow/internal/engine"
	"ragflow/internal/entity"
	"ragflow/internal/permission"
	permissionresponse "ragflow/internal/permission/response"
)

const (
	connectorInputTypePoll   = "poll"
	connectorStatusUnstarted = "0"
	defaultConnectorFreq     = 5
	defaultConnectorTimeout  = 60 * 29
	webFlowTTL               = 15 * time.Minute
	googleOAuthAuthorizeURL  = "https://accounts.google.com/o/oauth2/auth"
	googleOAuthTokenURL      = "https://oauth2.googleapis.com/token"
	boxOAuthAuthorizeURL     = "https://account.box.com/api/oauth2/authorize"
	boxOAuthTokenURL         = "https://api.box.com/oauth2/token"
	googleOAuthHTTPTimeout   = 7 * time.Second
)

var (
	googleDriveOAuthScopes = []string{
		"https://www.googleapis.com/auth/drive.readonly",
		"https://www.googleapis.com/auth/drive.metadata.readonly",
		"https://www.googleapis.com/auth/admin.directory.group.readonly",
		"https://www.googleapis.com/auth/admin.directory.user.readonly",
	}
	gmailOAuthScopes = []string{
		"https://www.googleapis.com/auth/gmail.readonly",
		"https://www.googleapis.com/auth/admin.directory.user.readonly",
		"https://www.googleapis.com/auth/admin.directory.group.readonly",
	}
	connectorRedisGet = kvrocks.Get
)

// Sentinel errors so handlers can map to the proper response codes.
var (
	// ErrConnectorNotFound is returned when a connector is not found.
	ErrConnectorNotFound = errors.New("can't find this Connector")
	// ErrConnectorNotBoundToKB is returned when the connector is not bound to the kb being rebuilt.
	ErrConnectorNotBoundToKB = errors.New("connector is not bound to this knowledge base")
	// ErrConnectorIDRequired is returned when a connector ID is missing.
	ErrConnectorIDRequired = errors.New("connector_id is required")
	// 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")
	// ErrConnectorInternal is a generic, safe-to-expose internal failure.
	ErrConnectorInternal = errors.New("Internal server error")
	// ErrConnectorEncryptedCredentials is returned when a request sends credentials in the stored "enc:" form.
	ErrConnectorEncryptedCredentials = errors.New("config.credentials must be plaintext, not an \"enc:\" value")
	// ErrConnectorInvalidFreq is returned when a connector frequency is not
	// positive. Handlers map it to a client error; callers match with errors.Is.
	ErrConnectorInvalidFreq = errors.New("connector frequency must be positive")
	// ErrConnectorInvalidName is returned when a connector name is empty or
	// wider than the connector.name column. Handlers map it to a client error.
	ErrConnectorInvalidName = errors.New("connector name is invalid")
)

// ConnectorService connector service
type ConnectorService struct {
	connectorDAO      *dao.ConnectorDAO
	connectorRegistry *syncerconnector.Registry
}

type syncTaskPublisher interface {
	PublishSyncerTask(taskID string) error
}

type syncTaskWakeupPublisher interface {
	PublishSyncerTaskWakeup(taskID string) error
}

type syncCheckpointLoader interface {
	LoadSyncCheckpoint(ctx context.Context, taskID string) (*syncerconnector.SyncCheckpointState, error)
}

type syncCheckpointDeleter interface {
	DeleteSyncCheckpoint(ctx context.Context, taskID string) error
}

var getSyncerTaskPublisher = func() (syncTaskPublisher, bool) {
	publisher, ok := engine.GetMessageQueueEngine().(syncTaskPublisher)
	return publisher, ok
}

var getSyncCheckpointLoader = func() (syncCheckpointLoader, bool) {
	loader, ok := engine.GetMessageQueueEngine().(syncCheckpointLoader)
	return loader, ok
}

var getSyncCheckpointDeleter = func() (syncCheckpointDeleter, bool) {
	deleter, ok := engine.GetMessageQueueEngine().(syncCheckpointDeleter)
	return deleter, ok
}

// NewConnectorService create connector service
func NewConnectorService() *ConnectorService {
	return &ConnectorService{
		connectorDAO:      dao.NewConnectorDAO(),
		connectorRegistry: newConnectorRegistry(),
	}
}

func newConnectorRegistry() *syncerconnector.Registry {
	registry := syncerconnector.NewRegistry()
	syncerconnector.RegisterBuiltIns(registry)
	return registry
}

// ListConnectorsResponse list connectors response
type ListConnectorsResponse struct {
	Connectors []*dao.ConnectorListItem `json:"connectors"`
}

// CreateConnectorRequest holds the fields used to create a connector.
type CreateConnectorRequest struct {
	Name        string         `json:"name"`
	Source      string         `json:"source"`
	Config      entity.JSONMap `json:"config"`
	RefreshFreq *int64         `json:"refresh_freq,omitempty"`
	PruneFreq   *int64         `json:"prune_freq,omitempty"`
	TimeoutSecs *int64         `json:"timeout_secs,omitempty"`
}

// RebuildConnectorRequest rebuild connector request.
type RebuildConnectorRequest struct {
	KbID string `json:"kb_id"`
}

// ResumeFailedSyncRequest resumes a failed connector sync task from checkpoint.
type ResumeFailedSyncRequest struct {
	KbID   string `json:"kb_id"`
	TaskID string `json:"task_id"`
}

type StartGoogleWebOAuthRequest struct {
	Credentials json.RawMessage `json:"credentials"`
	RedirectURI string          `json:"redirect_uri,omitempty"`
}

type StartGoogleWebOAuthResponse struct {
	FlowID           string `json:"flow_id"`
	AuthorizationURL string `json:"authorization_url"`
	ExpiresIn        int64  `json:"expires_in"`
}

type PollGoogleWebOAuthResultRequest struct {
	FlowID string `json:"flow_id"`
}

type PollGoogleWebOAuthResultResponse struct {
	Credentials string `json:"credentials"`
}

type googleWebOAuthState struct {
	UserID       string                 `json:"user_id"`
	ClientConfig map[string]interface{} `json:"client_config"`
	RedirectURI  string                 `json:"redirect_uri"`
	CodeVerifier string                 `json:"code_verifier"`
	CreatedAt    int64                  `json:"created_at"`
}

type googleWebOAuthResult struct {
	UserID      string `json:"user_id"`
	Credentials string `json:"credentials"`
}

type googleOAuthTokenResponse struct {
	AccessToken  string `json:"access_token"`
	RefreshToken string `json:"refresh_token,omitempty"`
	TokenType    string `json:"token_type,omitempty"`
	ExpiresIn    int64  `json:"expires_in,omitempty"`
	Scope        string `json:"scope,omitempty"`
	IDToken      string `json:"id_token,omitempty"`
	Error        string `json:"error,omitempty"`
	ErrorDesc    string `json:"error_description,omitempty"`
}

type googleOAuthCredentials struct {
	Token        string   `json:"token"`
	RefreshToken string   `json:"refresh_token,omitempty"`
	TokenURI     string   `json:"token_uri"`
	ClientID     string   `json:"client_id"`
	ClientSecret string   `json:"client_secret"`
	Scopes       []string `json:"scopes"`
	Expiry       string   `json:"expiry,omitempty"`
}

type boxWebOAuthState struct {
	UserID       string `json:"user_id"`
	AuthURL      string `json:"auth_url"`
	ClientID     string `json:"client_id"`
	ClientSecret string `json:"client_secret"`
	RedirectURI  string `json:"redirect_uri"`
	CreatedAt    int64  `json:"created_at"`
}

type boxWebOAuthCredentials struct {
	UserID       string `json:"user_id,omitempty"`
	ClientID     string `json:"client_id"`
	ClientSecret string `json:"client_secret"`
	AccessToken  string `json:"access_token"`
	RefreshToken string `json:"refresh_token"`
}

type StartBoxWebOAuthRequest struct {
	ClientID     string `json:"client_id"`
	ClientSecret string `json:"client_secret"`
	RedirectURI  string `json:"redirect_uri,omitempty"`
}

type StartBoxWebOAuthResponse struct {
	FlowID           string `json:"flow_id"`
	AuthorizationURL string `json:"authorization_url"`
	ExpiresIn        int64  `json:"expires_in"`
}

type PollBoxWebOAuthResultRequest struct {
	FlowID string `json:"flow_id"`
}

type PollBoxWebOAuthResultResponse struct {
	Credentials boxWebOAuthCredentials `json:"credentials"`
}

type boxOAuthTokenResponse struct {
	AccessToken  string `json:"access_token"`
	RefreshToken string `json:"refresh_token"`
	TokenType    string `json:"token_type,omitempty"`
	ExpiresIn    int64  `json:"expires_in,omitempty"`
	RestrictedTo any    `json:"restricted_to,omitempty"`
	Error        string `json:"error,omitempty"`
	ErrorDesc    string `json:"error_description,omitempty"`
}

// cancelConnectorTasks Stop connector tasks
func (s *ConnectorService) cancelConnectorTasks(ctx context.Context, connectorID string) error {
	if err := s.connectorDAO.CancelRunningOrScheduledLogs(ctx, dao.DB, connectorID); err != nil {
		return err
	}
	return s.connectorDAO.UpdateByID(ctx, dao.DB, connectorID, map[string]interface{}{"status": string(entity.TaskStatusCancel)})
}

// CreateConnector creates a connector owned by the current user.
func (s *ConnectorService) CreateConnector(ctx context.Context, req *CreateConnectorRequest) (*entity.Connector, error) {
	userID := permission.UserIDFromContext(ctx)
	if userID == "" {
		return nil, permission.ErrUnauthenticated
	}
	tenantID, err := requireCurrentTenantID(ctx)
	if err != nil {
		return nil, err
	}
	if err := permission.NewDatabaseChecker(dao.DB).CheckFeature(ctx, "file", permission.ActionWrite); err != nil {
		return nil, err
	}
	name := strings.TrimSpace(req.Name)
	if err := common.ValidateNameRunes(name, common.NameLimit128); err != nil {
		return nil, fmt.Errorf("%w: %v", ErrConnectorInvalidName, err)
	}
	req.Name = name
	if req.RefreshFreq != nil && *req.RefreshFreq <= 0 {
		return nil, fmt.Errorf("%w: refresh_freq got %d", ErrConnectorInvalidFreq, *req.RefreshFreq)
	}
	if req.PruneFreq != nil && *req.PruneFreq <= 0 {
		return nil, fmt.Errorf("%w: prune_freq got %d", ErrConnectorInvalidFreq, *req.PruneFreq)
	}
	if common.HasEncryptedConnectorCredentials(req.Config) {
		return nil, ErrConnectorEncryptedCredentials
	}
	refreshFreq := int64(defaultConnectorFreq)
	if req.RefreshFreq != nil {
		refreshFreq = *req.RefreshFreq
	}

	pruneFreq := int64(defaultConnectorFreq)
	if req.PruneFreq != nil {
		pruneFreq = *req.PruneFreq
	}

	timeoutSecs := int64(defaultConnectorTimeout)
	if req.TimeoutSecs != nil {
		timeoutSecs = *req.TimeoutSecs
	}

	connector := &entity.Connector{
		ID:          common.GenerateUUID(),
		UserID:      userID,
		TenantID:    tenantID,
		Name:        req.Name,
		Source:      req.Source,
		InputType:   connectorInputTypePoll,
		Config:      entity.ConnectorConfig(req.Config),
		RefreshFreq: refreshFreq,
		PruneFreq:   pruneFreq,
		TimeoutSecs: timeoutSecs,
		Status:      connectorStatusUnstarted,
	}

	if err := dao.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		if err := s.connectorDAO.Create(ctx, tx, connector); err != nil {
			return err
		}
		return permission.NewDatabaseChecker(tx).GrantResourceOwner(ctx, permission.ResourceRef{Kind: permission.ResourceKindConnector, ID: connector.ID})
	}); err != nil {
		return nil, err
	}

	created, err := s.connectorDAO.GetByID(ctx, dao.DB, connector.ID)
	if err != nil {
		return nil, err
	}
	return withDecryptedCredentials(created)
}

// GetConnector returns one connector when the user has read permission.
func (s *ConnectorService) GetConnector(ctx context.Context, connectorID string) (*entity.Connector, error) {
	if strings.TrimSpace(connectorID) == "" {
		return nil, ErrConnectorIDRequired
	}
	if err := permission.NewDatabaseChecker(dao.DB).CheckResource(ctx, permission.ResourceRef{Kind: permission.ResourceKindConnector, ID: connectorID}, permission.OperationRead); err != nil {
		return nil, err
	}

	connector, err := s.connectorDAO.GetByID(ctx, dao.DB, connectorID)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, ErrConnectorNotFound
		}
		common.Error("get connector failed", err, zap.String("connector_id", connectorID))
		return nil, ErrConnectorInternal
	}

	return withDecryptedCredentials(connector)
}

// withDecryptedCredentials decrypts the credentials of a connector that goes
// back to the client. The web form sends this config back on update.
func withDecryptedCredentials(connector *entity.Connector) (*entity.Connector, error) {
	config, err := common.DecryptConnectorCredentials(connector.Config)
	if err != nil {
		return nil, err
	}
	connector.Config = config
	return connector, nil
}

// ListConnectors returns connectors belonging to a tenant the caller can use.
func (s *ConnectorService) ListConnectors(ctx context.Context, requestedTenantID string) (*ListConnectorsResponse, error) {
	userID := strings.TrimSpace(permission.UserIDFromContext(ctx))
	if userID == "" {
		return nil, permission.ErrUnauthenticated
	}

	tenantID := strings.TrimSpace(requestedTenantID)
	if tenantID == "" {
		var err error
		tenantID, err = requireCurrentTenantID(ctx)
		if errors.Is(err, permission.ErrMembershipNotFound) {
			return &ListConnectorsResponse{Connectors: []*dao.ConnectorListItem{}}, nil
		}
		if err != nil {
			return nil, err
		}
	} else if err := permission.NewDatabaseChecker(dao.DB).CheckTenant(ctx, tenantID, permission.TenantMember); err != nil {
		return nil, err
	}
	connectors, err := s.connectorDAO.ListByTenantID(ctx, dao.DB, tenantID)
	if err != nil {
		return nil, err
	}

	return &ListConnectorsResponse{
		Connectors: connectors,
	}, nil
}

// TestConnector validates connector settings without persisting or syncing.
func (s *ConnectorService) TestConnector(ctx context.Context, connectorID string, config entity.JSONMap) error {
	var storedConnector *entity.Connector
	if connectorID != "" {
		permissionErr := permission.NewDatabaseChecker(dao.DB).CheckResource(ctx, permission.ResourceRef{Kind: permission.ResourceKindConnector, ID: connectorID}, permission.OperationUse)
		if permissionErr != nil && !errors.Is(permissionErr, permission.ErrResourceNotFound) {
			return permissionErr
		}
		if errors.Is(permissionErr, permission.ErrResourceNotFound) {
			if config == nil {
				return ErrConnectorNotFound
			}
			// A caller may test settings for an unsaved connector by supplying a
			// request config, but must not use that path to test an existing
			// connector they cannot access.
			if _, lookupErr := s.connectorDAO.GetByID(ctx, dao.DB, connectorID); lookupErr == nil {
				return ErrConnectorNotFound
			} else if !errors.Is(lookupErr, gorm.ErrRecordNotFound) {
				return lookupErr
			}
		} else {
			loadedConnector, err := s.connectorDAO.GetByID(ctx, dao.DB, connectorID)
			if err != nil {
				if errors.Is(err, gorm.ErrRecordNotFound) {
					return ErrConnectorNotFound
				}
				return err
			}
			storedConnector = loadedConnector
		}
	}

	source, connectorConfig, err := testConnectorSettings(storedConnector, config)
	if err != nil {
		return err
	}
	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)
	if !ok {
		return ErrConnectorTestUnsupported
	}
	return wrapConnectorValidationError(validator.ValidateConnectorSetting(ctx, connectorConfig))
}

func wrapConnectorValidationError(err error) error {
	if err == nil {
		return nil
	}
	var (
		valErr  *syncerconnector.ConnectorValidationError
		credErr *syncerconnector.ConnectorMissingCredentialError
		rateErr *syncerconnector.RateLimitTriedTooManyTimesError
	)
	if errors.As(err, &valErr) || errors.As(err, &credErr) || errors.As(err, &rateErr) {
		return err
	}
	return &syncerconnector.ConnectorValidationError{Message: err.Error()}
}

func testConnectorSettings(stored *entity.Connector, request entity.JSONMap) (string, entity.JSONMap, error) {
	source := ""
	var config entity.JSONMap
	if stored != nil {
		source = strings.TrimSpace(stored.Source)
		config = entity.JSONMap(stored.Config)
	}
	fromRequest := false
	if request != nil {
		if value := strings.TrimSpace(stringConfigValue(request["source"])); value != "" {
			source = value
		}
		if nested, ok := request["config"]; ok {
			config = jsonMapValue(nested)
			fromRequest = true
		} else if _, ok := request["source"]; !ok {
			config = request
			fromRequest = true
		}
	}
	if source == "" {
		return "", nil, fmt.Errorf("connector source is required")
	}
	if config == nil {
		return "", nil, fmt.Errorf("connector configuration is missing")
	}
	if fromRequest {
		if common.HasEncryptedConnectorCredentials(config) {
			return "", nil, &syncerconnector.ConnectorValidationError{Message: ErrConnectorEncryptedCredentials.Error()}
		}
		return source, config, nil
	}
	// Decrypt only the stored fallback, so testing new credentials works without the key.
	decrypted, err := common.DecryptConnectorCredentials(config)
	if err != nil {
		return "", nil, err
	}
	return source, decrypted, nil
}

func jsonMapValue(value any) entity.JSONMap {
	switch typed := value.(type) {
	case entity.JSONMap:
		return typed
	case map[string]any:
		return entity.JSONMap(typed)
	default:
		return nil
	}
}

func stringConfigValue(value any) string {
	switch typed := value.(type) {
	case string:
		return typed
	case nil:
		return ""
	default:
		return fmt.Sprint(typed)
	}
}

func (s *ConnectorService) StartGoogleWebOAuth(ctx context.Context, source string, req *StartGoogleWebOAuthRequest) (*StartGoogleWebOAuthResponse, common.ErrorCode, error) {
	userID := strings.TrimSpace(permission.UserIDFromContext(ctx))
	if userID == "" {
		return nil, common.CodeUnauthorized, permission.ErrUnauthenticated
	}
	source = strings.TrimSpace(source)
	if source == "" {
		source = "google-drive"
	}
	if source != "google-drive" && source != "gmail" {
		return nil, common.CodeArgumentError, fmt.Errorf("invalid Google OAuth type")
	}

	if req == nil || len(req.Credentials) == 0 {
		return nil, common.CodeArgumentError, fmt.Errorf("required argument is missing: credentials")
	}

	redirectURI := strings.TrimSpace(req.RedirectURI)
	if redirectURI == "" {
		redirectURI = defaultGoogleWebOAuthRedirectURI(source)
	}
	if redirectURI == "" {
		return nil, common.CodeServerError, fmt.Errorf("not configure Google OAuth redirect URI on the server")
	}

	credentials, err := loadGoogleCredentials(req.Credentials)
	if err != nil {
		return nil, common.CodeArgumentError, err
	}
	if hasRefreshToken(credentials) {
		return nil, common.CodeArgumentError, fmt.Errorf("uploaded credentials already include a refresh token")
	}

	clientConfig, err := getGoogleWebClientConfig(credentials)
	if err != nil {
		return nil, common.CodeArgumentError, err
	}

	webConfig, _ := clientConfig["web"].(map[string]interface{})
	clientID := strings.TrimSpace(stringValue(webConfig["client_id"]))
	authURI := strings.TrimSpace(stringValue(webConfig["auth_uri"]))
	if authURI == "" {
		authURI = googleOAuthAuthorizeURL
	}
	if clientID == "" || authURI == "" {
		return nil, common.CodeServerError, fmt.Errorf("failed to initialize Google OAuth flow. Please verify the uploaded client configuration")
	}

	codeVerifier, codeChallenge, err := newPKCEChallenge()
	if err != nil {
		return nil, common.CodeServerError, err
	}

	flowID := common.GenerateUUID()
	authorizationURL, err := buildGoogleAuthorizationURL(authURI, clientID, redirectURI, flowID, googleOAuthScopesForSource(source), codeChallenge)
	if err != nil {
		return nil, common.CodeServerError, fmt.Errorf("failed to initialize Google OAuth flow. Please verify the uploaded client configuration")
	}

	redisClient := kvrocks.Get()
	if redisClient == nil {
		return nil, common.CodeServerError, fmt.Errorf("no configure Redis on the server")
	}

	state := googleWebOAuthState{
		UserID:       userID,
		ClientConfig: clientConfig,
		RedirectURI:  redirectURI,
		CodeVerifier: codeVerifier,
		CreatedAt:    time.Now().Unix(),
	}
	if ok := redisClient.SetObj(ctx, webStateCacheKey(flowID, source), state, webFlowTTL); !ok {
		return nil, common.CodeServerError, fmt.Errorf("failed to initialize Google OAuth flow. Please verify the uploaded client configuration")
	}

	return &StartGoogleWebOAuthResponse{
		FlowID:           flowID,
		AuthorizationURL: authorizationURL,
		ExpiresIn:        int64(webFlowTTL.Seconds()),
	}, common.CodeSuccess, nil
}

func (s *ConnectorService) GoogleWebOAuthCallback(ctx context.Context, source, stateID, oauthError, errorDescription, code string) string {
	source = strings.TrimSpace(source)
	if source != "google-drive" && source != "gmail" {
		return renderWebOAuthPopup("", false, "Invalid Google OAuth type.", source)
	}

	stateID = strings.TrimSpace(stateID)
	if stateID == "" {
		return renderWebOAuthPopup("", false, "Missing OAuth state parameter.", source)
	}

	redisClient := kvrocks.Get()
	if redisClient == nil {
		return renderWebOAuthPopup(stateID, false, "Authorization session expired. Please restart from the main window.", source)
	}

	stateKey := webStateCacheKey(stateID, source)
	var state googleWebOAuthState
	if ok := redisClient.GetObj(ctx, stateKey, &state); !ok {
		return renderWebOAuthPopup(stateID, false, "Authorization session expired. Please restart from the main window.", source)
	}

	if state.ClientConfig == nil {
		redisClient.Delete(ctx, stateKey)
		return renderWebOAuthPopup(stateID, false, "Authorization session was invalid. Please retry.", source)
	}

	if strings.TrimSpace(oauthError) != "" {
		redisClient.Delete(ctx, stateKey)
		message := strings.TrimSpace(errorDescription)
		if message == "" {
			message = strings.TrimSpace(oauthError)
		}
		if message == "" {
			message = "Authorization was cancelled."
		}
		return renderWebOAuthPopup(stateID, false, message, source)
	}

	code = strings.TrimSpace(code)
	if code == "" {
		return renderWebOAuthPopup(stateID, false, "Missing authorization code from Google.", source)
	}

	credentials, err := exchangeGoogleWebOAuthCode(state.ClientConfig, googleOAuthScopesForSource(source), state.RedirectURI, code, state.CodeVerifier)
	if err != nil {
		redisClient.Delete(ctx, stateKey)
		return renderWebOAuthPopup(stateID, false, "Failed to exchange tokens with Google. Please retry.", source)
	}

	result := googleWebOAuthResult{
		UserID:      state.UserID,
		Credentials: credentials,
	}
	if ok := redisClient.SetObj(ctx, webResultCacheKey(stateID, source), result, webFlowTTL); !ok {
		redisClient.Delete(ctx, stateKey)
		return renderWebOAuthPopup(stateID, false, "Failed to exchange tokens with Google. Please retry.", source)
	}
	redisClient.Delete(ctx, stateKey)

	return renderWebOAuthPopup(stateID, true, "Authorization completed successfully.", source)
}

func (s *ConnectorService) PollGoogleWebOAuthResult(ctx context.Context, source string, req *PollGoogleWebOAuthResultRequest) (*PollGoogleWebOAuthResultResponse, common.ErrorCode, error) {
	userID := strings.TrimSpace(permission.UserIDFromContext(ctx))
	if userID == "" {
		return nil, common.CodeUnauthorized, permission.ErrUnauthenticated
	}
	source = strings.TrimSpace(source)
	if source != "google-drive" && source != "gmail" {
		return nil, common.CodeArgumentError, fmt.Errorf("invalid Google OAuth type")
	}
	if req == nil || strings.TrimSpace(req.FlowID) == "" {
		return nil, common.CodeArgumentError, fmt.Errorf("required argument is missing: flow_id")
	}

	redisClient := kvrocks.Get()
	if redisClient == nil {
		return nil, common.CodeRunning, fmt.Errorf("authorization is still pending")
	}

	resultKey := webResultCacheKey(strings.TrimSpace(req.FlowID), source)
	var result googleWebOAuthResult
	if ok := redisClient.GetObj(ctx, resultKey, &result); !ok {
		return nil, common.CodeRunning, fmt.Errorf("authorization is still pending")
	}

	if result.UserID != userID {
		return nil, common.CodePermissionError, fmt.Errorf("you are not allowed to access this authorization result")
	}

	redisClient.Delete(ctx, resultKey)
	return &PollGoogleWebOAuthResultResponse{Credentials: result.Credentials}, common.CodeSuccess, nil
}

func defaultGoogleWebOAuthRedirectURI(source string) string {
	if source == "gmail" {
		return getEnvDefault(common.EnvGmailWebOAuthRedirectURI, "http://localhost:9380/api/v1/connectors/gmail/oauth/web/callback")
	}
	return getEnvDefault(common.EnvGoogleDriveWebOAuthRedirectURI, "http://localhost:9380/api/v1/connectors/google-drive/oauth/web/callback")
}

func getEnvDefault(key, fallback string) string {
	if value := strings.TrimSpace(common.GetEnv(key)); value != "" {
		return value
	}
	return fallback
}

func webStateCacheKey(flowID, source string) string {
	return fmt.Sprintf("%s_web_flow_state:%s", source, flowID)
}

func webResultCacheKey(flowID, source string) string {
	return fmt.Sprintf("%s_web_flow_result:%s", source, flowID)
}

func loadGoogleCredentials(raw json.RawMessage) (map[string]interface{}, error) {
	var credentials map[string]interface{}
	if err := json.Unmarshal(raw, &credentials); err == nil && credentials != nil {
		return credentials, nil
	}

	var rawString string
	if err := json.Unmarshal(raw, &rawString); err != nil {
		return nil, fmt.Errorf("invalid Google credentials JSON")
	}
	if err := json.Unmarshal([]byte(strings.TrimSpace(rawString)), &credentials); err != nil || credentials == nil {
		return nil, fmt.Errorf("invalid Google credentials JSON")
	}
	return credentials, nil
}

func hasRefreshToken(credentials map[string]interface{}) bool {
	value, ok := credentials["refresh_token"]
	if !ok || value == nil {
		return false
	}
	if token, ok := value.(string); ok {
		return strings.TrimSpace(token) != ""
	}
	return true
}

func getGoogleWebClientConfig(credentials map[string]interface{}) (map[string]interface{}, error) {
	webSection, ok := credentials["web"].(map[string]interface{})
	if !ok || webSection == nil {
		return nil, fmt.Errorf("google OAuth JSON must include a 'web' client configuration to use browser-based authorization")
	}
	return map[string]interface{}{"web": webSection}, nil
}

func stringValue(value interface{}) string {
	if s, ok := value.(string); ok {
		return s
	}
	return ""
}

func googleOAuthScopesForSource(source string) []string {
	if source == "gmail" {
		return gmailOAuthScopes
	}
	return googleDriveOAuthScopes
}

func newPKCEChallenge() (string, string, error) {
	randomBytes := make([]byte, 64)
	if _, err := rand.Read(randomBytes); err != nil {
		return "", "", fmt.Errorf("failed to generate OAuth code verifier: %w", err)
	}
	verifier := base64.RawURLEncoding.EncodeToString(randomBytes)
	sum := sha256.Sum256([]byte(verifier))
	challenge := base64.RawURLEncoding.EncodeToString(sum[:])
	return verifier, challenge, nil
}

func buildGoogleAuthorizationURL(authURI, clientID, redirectURI, state string, scopes []string, codeChallenge string) (string, error) {
	parsedURL, err := url.Parse(authURI)
	if err != nil {
		return "", err
	}

	query := parsedURL.Query()
	query.Set("client_id", clientID)
	query.Set("redirect_uri", redirectURI)
	query.Set("response_type", "code")
	query.Set("scope", strings.Join(scopes, " "))
	query.Set("access_type", "offline")
	query.Set("include_granted_scopes", "true")
	query.Set("prompt", "consent")
	query.Set("state", state)
	query.Set("code_challenge", codeChallenge)
	query.Set("code_challenge_method", "S256")
	parsedURL.RawQuery = query.Encode()
	return parsedURL.String(), nil
}

func exchangeGoogleWebOAuthCode(clientConfig map[string]interface{}, scopes []string, redirectURI, code, codeVerifier string) (string, error) {
	webConfig, ok := clientConfig["web"].(map[string]interface{})
	if !ok || webConfig == nil {
		return "", fmt.Errorf("invalid Google OAuth client configuration")
	}

	clientID := strings.TrimSpace(stringValue(webConfig["client_id"]))
	clientSecret := strings.TrimSpace(stringValue(webConfig["client_secret"]))
	configuredTokenURI := strings.TrimSpace(stringValue(webConfig["token_uri"]))
	if configuredTokenURI != "" && configuredTokenURI != googleOAuthTokenURL {
		return "", fmt.Errorf("invalid Google OAuth client configuration: token_uri must use Google's token endpoint")
	}
	if clientID == "" {
		return "", fmt.Errorf("invalid Google OAuth client configuration")
	}
	tokenURI := googleOAuthTokenURL

	form := url.Values{}
	form.Set("client_id", clientID)
	if clientSecret != "" {
		form.Set("client_secret", clientSecret)
	}
	form.Set("code", code)
	form.Set("redirect_uri", redirectURI)
	form.Set("grant_type", "authorization_code")
	if strings.TrimSpace(codeVerifier) != "" {
		form.Set("code_verifier", codeVerifier)
	}

	ctx, cancel := context.WithTimeout(context.Background(), googleOAuthHTTPTimeout)
	defer cancel()

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, tokenURI, strings.NewReader(form.Encode()))
	if err != nil {
		return "", err
	}
	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
	req.Header.Set("Accept", "application/json")

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()

	body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	if err != nil {
		return "", err
	}

	var token googleOAuthTokenResponse
	if err := json.Unmarshal(body, &token); err != nil {
		return "", err
	}
	if resp.StatusCode >= http.StatusBadRequest || token.Error != "" {
		if token.ErrorDesc != "" {
			return "", errors.New(token.ErrorDesc)
		}
		if token.Error != "" {
			return "", errors.New(token.Error)
		}
		return "", fmt.Errorf("google token exchange failed: HTTP %d", resp.StatusCode)
	}
	if token.AccessToken == "" {
		return "", fmt.Errorf("google token exchange failed: empty access_token")
	}

	expiry := ""
	if token.ExpiresIn > 0 {
		expiry = time.Now().Add(time.Duration(token.ExpiresIn) * time.Second).UTC().Format(time.RFC3339Nano)
	}
	credentials := googleOAuthCredentials{
		Token:        token.AccessToken,
		RefreshToken: token.RefreshToken,
		TokenURI:     tokenURI,
		ClientID:     clientID,
		ClientSecret: clientSecret,
		Scopes:       scopes,
		Expiry:       expiry,
	}
	data, err := json.Marshal(credentials)
	if err != nil {
		return "", err
	}
	return string(data), nil
}

func renderWebOAuthPopup(flowID string, success bool, message, source string) string {
	status := "error"
	autoClose := ""
	if success {
		status = "success"
		autoClose = "window.close();"
	}
	payloadType := fmt.Sprintf("ragflow-%s-oauth", source)
	payload, _ := json.Marshal(map[string]string{
		"type":    payloadType,
		"status":  status,
		"flowId":  flowID,
		"message": message,
	})

	title := fmt.Sprintf("%s Authorization", webOAuthSourceDisplayName(source))
	heading := "Authorization failed"
	if success {
		heading = "Authorization complete"
	}

	return fmt.Sprintf(`<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="utf-8" />
  <title>%s</title>
  <style>
    body {
      font-family: Arial, sans-serif;
      background: #f8fafc;
      color: #0f172a;
      display: flex;
      flex-direction: column;
      align-items: center;
      justify-content: center;
      min-height: 100vh;
      margin: 0;
    }
    .card {
      background: white;
      padding: 32px;
      border-radius: 12px;
      box-shadow: 0 8px 30px rgba(15, 23, 42, 0.1);
      max-width: 420px;
      text-align: center;
    }
    h1 {
      font-size: 1.5rem;
      margin-bottom: 12px;
    }
    p {
      font-size: 0.95rem;
      line-height: 1.5;
    }
  </style>
</head>
<body>
  <div class="card">
    <h1>%s</h1>
    <p>%s</p>
    <p>You can close this window.</p>
  </div>
  <script>
    (function(){
      if (window.opener) {
        window.opener.postMessage(%s, "*");
      }
      %s
    })();
  </script>
</body>
</html>`, html.EscapeString(title), html.EscapeString(heading), html.EscapeString(message), string(payload), autoClose)
}

func webOAuthSourceDisplayName(source string) string {
	if source == "gmail" {
		return "Gmail"
	}
	if source == "google-drive" {
		return "Google Drive"
	}
	if source == "box" {
		return "Box"
	}
	return "OAuth"
}

func (s *ConnectorService) DeleteConnector(ctx context.Context, connectorID string) (bool, common.ErrorCode, error) {
	if connectorID == "" {
		return false, common.CodeDataError, fmt.Errorf("connector_id is required")
	}
	if err := permission.NewDatabaseChecker(dao.DB).CheckResource(ctx, permission.ResourceRef{Kind: permission.ResourceKindConnector, ID: connectorID}, permission.OperationDelete); err != nil {
		code, permissionErr := permissionresponse.NormalizeHidden(err)
		return false, code, permissionErr
	}

	connector, err := s.connectorDAO.GetByID(ctx, dao.DB, connectorID)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return false, common.CodeDataError, fmt.Errorf("can't find this Connector")
		}
		return false, common.CodeServerError, err
	}

	if err = s.cancelConnectorTasks(ctx, connector.ID); err != nil {
		return false, common.CodeServerError, err
	}

	if err = s.connectorDAO.DeleteByID(ctx, dao.DB, connector.ID); err != nil {
		return false, common.CodeServerError, err
	}
	return true, common.CodeSuccess, nil
}

type UpdateConnectorRequest struct {
	PruneFreq   *int64         `json:"prune_freq,omitempty"`
	RefreshFreq *int64         `json:"refresh_freq,omitempty"`
	Config      entity.JSONMap `json:"config,omitempty"`
	TimeoutSecs *int64         `json:"timeout_secs,omitempty"`
	Reschedule  bool           `json:"reschedule,omitempty"`
	Status      string         `json:"status,omitempty"`
}

func (s *ConnectorService) UpdateConnector(ctx context.Context, connectorID string, req *UpdateConnectorRequest) (*entity.Connector, common.ErrorCode, error) {
	if connectorID == "" {
		return nil, common.CodeDataError, fmt.Errorf("connector_id is required")
	}
	if err := permission.NewDatabaseChecker(dao.DB).CheckResource(ctx, permission.ResourceRef{Kind: permission.ResourceKindConnector, ID: connectorID}, permission.OperationUpdate); err != nil {
		code, permissionErr := permissionresponse.NormalizeHidden(err)
		return nil, code, permissionErr
	}
	if req != nil && (req.Reschedule || isConnectorCancelStatus(req.Status) || isConnectorScheduleStatus(req.Status)) {
		if err := permission.NewDatabaseChecker(dao.DB).CheckResource(ctx, permission.ResourceRef{Kind: permission.ResourceKindConnector, ID: connectorID}, permission.OperationRun); err != nil {
			code, permissionErr := permissionresponse.NormalizeHidden(err)
			return nil, code, permissionErr
		}
	}

	connector, err := s.connectorDAO.GetByID(ctx, dao.DB, connectorID)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, common.CodeDataError, fmt.Errorf("can't find this Connector")
		}
		return nil, common.CodeServerError, err
	}

	// The response returns the stored credentials. Fail before any write when they
	// cannot be decrypted; a request with new credentials replaces them instead.
	if req == nil || req.Config == nil {
		if _, err = common.DecryptConnectorCredentials(connector.Config); err != nil {
			return nil, common.CodeServerError, err
		}
	}

	updates := map[string]interface{}{}
	if req != nil {
		if req.PruneFreq != nil {
			if *req.PruneFreq <= 0 {
				return nil, common.CodeDataError, fmt.Errorf("%w: prune_freq got %d", ErrConnectorInvalidFreq, *req.PruneFreq)
			}
			updates["prune_freq"] = *req.PruneFreq
		}
		if req.RefreshFreq != nil {
			if *req.RefreshFreq <= 0 {
				return nil, common.CodeDataError, fmt.Errorf("%w: refresh_freq got %d", ErrConnectorInvalidFreq, *req.RefreshFreq)
			}
			updates["refresh_freq"] = *req.RefreshFreq
		}
		if req.Config != nil {
			if common.HasEncryptedConnectorCredentials(req.Config) {
				return nil, common.CodeDataError, ErrConnectorEncryptedCredentials
			}
			// GORM writes map update values as given, so the value itself must be the type that encrypts.
			updates["config"] = entity.ConnectorConfig(req.Config)
		}
		if req.TimeoutSecs != nil {
			updates["timeout_secs"] = *req.TimeoutSecs
		}
	}

	if len(updates) > 0 {
		if err = s.connectorDAO.UpdateByID(ctx, dao.DB, connectorID, updates); err != nil {
			return nil, common.CodeServerError, err
		}
	}

	if req != nil {
		if req.Reschedule {
			if err = s.cancelConnectorTasks(ctx, connectorID); err != nil {
				return nil, common.CodeServerError, err
			}
			taskIDs, err := s.connectorDAO.ScheduleConnectorTasks(ctx, dao.DB, connectorID)
			if err != nil {
				return nil, common.CodeServerError, err
			}
			publishSyncerTasks(taskIDs)
		} else if isConnectorCancelStatus(req.Status) {
			if err = s.cancelConnectorTasks(ctx, connectorID); err != nil {
				return nil, common.CodeServerError, err
			}
		} else if isConnectorScheduleStatus(req.Status) {
			taskIDs, err := s.connectorDAO.ScheduleConnectorTasks(ctx, dao.DB, connectorID)
			if err != nil {
				return nil, common.CodeServerError, err
			}
			publishSyncerTasks(taskIDs)
		}
	}

	connector, err = s.connectorDAO.GetByID(ctx, dao.DB, connectorID)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, common.CodeDataError, fmt.Errorf("can't find this Connector")
		}
		return nil, common.CodeServerError, err
	}

	connector, err = withDecryptedCredentials(connector)
	if err != nil {
		return nil, common.CodeServerError, err
	}
	return connector, common.CodeSuccess, nil
}

func isConnectorCancelStatus(status string) bool {
	status = strings.TrimSpace(status)
	return status == string(entity.TaskStatusCancel) || strings.EqualFold(status, "CANCEL")
}

func isConnectorScheduleStatus(status string) bool {
	status = strings.TrimSpace(status)
	return status == string(entity.TaskStatusSchedule) || strings.EqualFold(status, "SCHEDULE")
}

// RebuildConnector schedules a rebuild for an accessible connector and knowledge base.
func (s *ConnectorService) RebuildConnector(ctx context.Context, connectorID, kbID string) (bool, common.ErrorCode, error) {
	if connectorID == "" {
		return false, common.CodeDataError, fmt.Errorf("connector_id is required")
	}
	if kbID == "" {
		return false, common.CodeArgumentError, fmt.Errorf("required argument is missing: kb_id")
	}
	if err := permission.NewDatabaseChecker(dao.DB).CheckResource(ctx, permission.ResourceRef{Kind: permission.ResourceKindConnector, ID: connectorID}, permission.OperationRun); err != nil {
		code, permissionErr := permissionresponse.NormalizeHidden(err)
		return false, code, permissionErr
	}

	connector, err := s.connectorDAO.GetByID(ctx, dao.DB, connectorID)
	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return false, common.CodeDataError, fmt.Errorf("can't find this Connector")
		}
		return false, common.CodeServerError, err
	}

	// The caller-supplied kb is targeted by delete + re-sync below, so it must
	// be accessible to the caller and the connector must be bound to it.
	if err := permission.NewDatabaseChecker(dao.DB).CheckResource(ctx, permission.ResourceRef{Kind: permission.ResourceKindDataset, ID: kbID}, permission.OperationRun); err != nil {
		common.Warn("rebuild denied: kb not accessible",
			zap.String("connector_id", connectorID), zap.String("kb_id", kbID), zap.String("user_id", permission.UserIDFromContext(ctx)))
		code, permissionErr := permissionresponse.Normalize(err)
		return false, code, permissionErr
	}
	bound, err := s.connectorDAO.Connector2KBExists(ctx, dao.DB, connectorID, kbID)
	if err != nil {
		return false, common.CodeServerError, err
	}
	if !bound {
		common.Warn("rebuild denied: connector not bound to kb",
			zap.String("connector_id", connectorID), zap.String("kb_id", kbID), zap.String("user_id", permission.UserIDFromContext(ctx)))
		return false, common.CodeAuthenticationError, ErrConnectorNotBoundToKB
	}

	sourceType := fmt.Sprintf("%s/%s", connector.Source, connector.ID)
	documents, err := s.connectorDAO.ListDocumentsByKBAndSourceType(ctx, dao.DB, kbID, sourceType)
	if err != nil {
		return false, common.CodeServerError, err
	}

	s.deleteConnectorDocumentChunks(ctx, connector.TenantID, kbID, documents)

	taskIDs, oldSyncTaskIDs, err := s.connectorDAO.RebuildConnector(ctx, dao.DB, connector, kbID, documents)
	if err != nil {
		return false, common.CodeServerError, err
	}

	if err = deleteSyncCheckpoints(ctx, oldSyncTaskIDs); err != nil {
		common.Warn("delete sync checkpoints failed during rebuild",
			zap.String("connector_id", connectorID), zap.Error(err))
	}

	publishSyncerTasks(taskIDs)
	return true, common.CodeSuccess, nil
}

// ResumeFailedSync schedules a failed sync task to continue from its saved checkpoint.
func (s *ConnectorService) ResumeFailedSync(ctx context.Context, connectorID string, req *ResumeFailedSyncRequest) (bool, common.ErrorCode, error) {
	if connectorID == "" {
		return false, common.CodeDataError, fmt.Errorf("connector_id is required")
	}
	if req == nil {
		return false, common.CodeArgumentError, fmt.Errorf("request is required")
	}
	// check KBid and TaskID
	req.KbID = strings.TrimSpace(req.KbID)
	req.TaskID = strings.TrimSpace(req.TaskID)
	if req.KbID == "" {
		return false, common.CodeArgumentError, fmt.Errorf("required argument is missing: kb_id")
	}
	if req.TaskID == "" {
		return false, common.CodeArgumentError, fmt.Errorf("required argument is missing: task_id")
	}
	if err := permission.NewDatabaseChecker(dao.DB).CheckResource(ctx, permission.ResourceRef{Kind: permission.ResourceKindConnector, ID: connectorID}, permission.OperationRun); err != nil {
		code, permissionErr := permissionresponse.NormalizeHidden(err)
		return false, code, permissionErr
	}
	connectorRef := permission.ResourceRef{Kind: permission.ResourceKindConnector, ID: connectorID}
	taskScope, err := permission.NewDatabaseChecker(dao.DB).Scope(ctx, permission.ScopeQuery{
		Kind:      permission.ResourceKindSyncTask,
		Parent:    connectorRef,
		Operation: permission.OperationRun,
		Entry:     &connectorRef,
		EntryOp:   permission.OperationRun,
	})
	if err != nil {
		code, permissionErr := permissionresponse.NormalizeHidden(err)
		return false, code, permissionErr
	}
	taskAccessible := false
	for _, taskID := range taskScope.ResourceIDs {
		if taskID == req.TaskID {
			taskAccessible = true
			break
		}
	}
	if !taskAccessible {
		code, permissionErr := permissionresponse.NormalizeHidden(permission.ErrResourceNotFound)
		return false, code, permissionErr
	}
	if err := permission.NewDatabaseChecker(dao.DB).CheckResource(ctx, permission.ResourceRef{Kind: permission.ResourceKindDataset, ID: req.KbID}, permission.OperationRun); err != nil {
		code, permissionErr := permissionresponse.NormalizeHidden(err)
		return false, code, permissionErr
	}
	// load the checkpoint
	loader, ok := getSyncCheckpointLoader()
	if !ok {
		return false, common.CodeServerError, fmt.Errorf("sync checkpoint store is not configured")
	}

	checkpoint, err := loader.LoadSyncCheckpoint(ctx, req.TaskID)
	if err != nil {
		return false, common.CodeServerError, err
	}
	if checkpoint == nil || checkpoint.Checkpoint == nil {
		return false, common.CodeDataError, fmt.Errorf("checkpoint not found, cannot resume failed sync task")
	}
	if checkpoint.TaskID != req.TaskID || checkpoint.ConnectorID != connectorID || checkpoint.KBID != req.KbID {
		return false, common.CodeDataError, fmt.Errorf("checkpoint does not match connector, knowledge base, and task")
	}
	// resume the task
	if err = s.connectorDAO.ResumeFailedSyncTask(ctx, dao.DB, connectorID, req.KbID, req.TaskID); err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return false, common.CodeDataError, fmt.Errorf("failed sync task not found")
		}
		return false, common.CodeServerError, err
	}
	if err = publishSyncerTaskWakeup(req.TaskID); err != nil {
		return false, common.CodeServerError, err
	}
	return true, common.CodeSuccess, nil
}

func publishSyncerTasks(taskIDs []string) {
	if len(taskIDs) == 0 {
		return
	}
	publisher, ok := getSyncerTaskPublisher()
	if !ok {
		common.Warn("syncer task publisher is not configured")
		return
	}
	for _, taskID := range taskIDs {
		if taskID == "" {
			continue
		}
		if err := publisher.PublishSyncerTask(taskID); err != nil {
			common.Warn("syncer task publish failed", zap.String("task_id", taskID), zap.Error(err))
		}
	}
}

// publishSyncerTaskWakeup publish syncer task wakeup message to nats
func publishSyncerTaskWakeup(taskID string) error {
	publisher, ok := getSyncerTaskPublisher()
	if !ok {
		return fmt.Errorf("syncer task publisher is not configured")
	}
	if wakeupPublisher, ok := publisher.(syncTaskWakeupPublisher); ok {
		return wakeupPublisher.PublishSyncerTaskWakeup(taskID)
	}
	return publisher.PublishSyncerTask(taskID)
}

func deleteSyncCheckpoints(ctx context.Context, taskIDs []string) error {
	if len(taskIDs) == 0 {
		return nil
	}
	deleter, ok := getSyncCheckpointDeleter()
	if !ok {
		return fmt.Errorf("sync checkpoint store is not configured")
	}
	for _, taskID := range taskIDs {
		if taskID == "" {
			continue
		}
		if err := deleter.DeleteSyncCheckpoint(ctx, taskID); err != nil {
			return err
		}
	}
	return nil
}

func (s *ConnectorService) deleteConnectorDocumentChunks(ctx context.Context, tenantID, kbID string, documents []*entity.Document) {
	docEngine := engine.Get()
	if docEngine == nil {
		return
	}

	indexName := fmt.Sprintf("ragflow_%s", tenantID)
	for _, document := range documents {
		_, _ = docEngine.DeleteChunks(ctx, map[string]interface{}{"doc_id": document.ID}, indexName, kbID)
	}
}

func (s *ConnectorService) ListLog(ctx context.Context, connectorID string, page, pageSize int) ([]*entity.ConnectorSyncLog, int64, common.ErrorCode, error) {
	if connectorID == "" {
		return nil, 0, common.CodeDataError, fmt.Errorf("connector_id is required")
	}

	checker := permission.NewDatabaseChecker(dao.DB)
	connectorRef := permission.ResourceRef{Kind: permission.ResourceKindConnector, ID: connectorID}
	if err := checker.CheckResource(ctx, connectorRef, permission.OperationRead); err != nil {
		code, permissionErr := permissionresponse.NormalizeHidden(err)
		return nil, 0, code, permissionErr
	}
	taskScope, err := checker.Scope(ctx, permission.ScopeQuery{
		Kind:      permission.ResourceKindSyncTask,
		Parent:    connectorRef,
		Operation: permission.OperationRead,
		Entry:     &connectorRef,
		EntryOp:   permission.OperationRead,
	})
	if err != nil {
		code, permissionErr := permissionresponse.NormalizeHidden(err)
		return nil, 0, code, permissionErr
	}
	taskIDs := taskScope.ResourceIDs
	if taskScope.Mode != permission.ScopeIDs {
		taskIDs = []string{}
	}

	if page < 1 {
		page = common.DefaultPage
	}
	if pageSize <= 0 || pageSize > common.MaxPageSize {
		pageSize = common.DefaultPageSize
	}
	offset := (page - 1) * pageSize

	logs, total, err := s.connectorDAO.ListLogsByTaskIDs(ctx, dao.DB, taskIDs, "", offset, pageSize)
	if err != nil {
		return nil, 0, common.CodeServerError, fmt.Errorf("failed to fetch connector logs: %w", err)
	}
	if logs == nil {
		logs = []*entity.ConnectorSyncLog{}
	}
	return logs, total, common.CodeSuccess, nil
}

// ListLogs lists sync logs for the current user with pagination.
// When datasetID is non-empty, only logs of that dataset are returned.
func (s *ConnectorService) ListLogs(ctx context.Context, datasetID string, page, pageSize int) ([]*entity.ConnectorSyncLog, int64, common.ErrorCode, error) {
	if strings.TrimSpace(permission.UserIDFromContext(ctx)) == "" {
		return nil, 0, common.CodeUnauthorized, permission.ErrUnauthenticated
	}

	if datasetID != "" {
		if err := permission.NewDatabaseChecker(dao.DB).CheckResource(ctx, permission.ResourceRef{Kind: permission.ResourceKindDataset, ID: datasetID}, permission.OperationRead); err != nil {
			code, permissionErr := permissionresponse.NormalizeHidden(err)
			return nil, 0, code, permissionErr
		}
	}
	checker := permission.NewDatabaseChecker(dao.DB)
	connectorScope, err := checker.Scope(ctx, permission.ScopeQuery{
		Kind:      permission.ResourceKindConnector,
		Operation: permission.OperationRead,
	})
	if err != nil {
		return nil, 0, common.CodeServerError, err
	}
	connectorIDs := connectorScope.ResourceIDs
	if connectorScope.Mode != permission.ScopeIDs {
		connectorIDs = []string{}
	}
	var taskIDs []string
	for _, connectorID := range connectorIDs {
		connectorRef := permission.ResourceRef{Kind: permission.ResourceKindConnector, ID: connectorID}
		taskScope, err := checker.Scope(ctx, permission.ScopeQuery{
			Kind:      permission.ResourceKindSyncTask,
			Parent:    connectorRef,
			Operation: permission.OperationRead,
			Entry:     &connectorRef,
			EntryOp:   permission.OperationRead,
		})
		if err != nil {
			code, permissionErr := permissionresponse.NormalizeHidden(err)
			return nil, 0, code, permissionErr
		}
		if taskScope.Mode == permission.ScopeIDs {
			taskIDs = append(taskIDs, taskScope.ResourceIDs...)
		}
	}

	offset, limit := 0, pageSize
	if pageSize <= 0 {
		// pageSize == 0 means no pagination: return every matching row.
		limit = 0
	} else {
		if page < 1 {
			page = common.DefaultPage
		}
		if pageSize > common.MaxPageSize {
			limit = common.DefaultPageSize
		}
		offset = (page - 1) * limit
	}

	logs, total, err := s.connectorDAO.ListLogsByTaskIDs(ctx, dao.DB, taskIDs, datasetID, offset, limit)
	if err != nil {
		return nil, 0, common.CodeServerError, fmt.Errorf("failed to fetch sync logs: %w", err)
	}
	if logs == nil {
		logs = []*entity.ConnectorSyncLog{}
	}
	return logs, total, common.CodeSuccess, nil
}

func (s *ConnectorService) StartBoxWebOAuth(ctx context.Context, req *StartBoxWebOAuthRequest) (*StartBoxWebOAuthResponse, common.ErrorCode, error) {
	userID := strings.TrimSpace(permission.UserIDFromContext(ctx))
	if userID == "" {
		return nil, common.CodeUnauthorized, permission.ErrUnauthenticated
	}
	var clientID, clientSecret, redirectURI string
	if req != nil {
		clientID = strings.TrimSpace(req.ClientID)
		clientSecret = strings.TrimSpace(req.ClientSecret)
		redirectURI = strings.TrimSpace(req.RedirectURI)
	}
	if clientID == "" || clientSecret == "" {
		return nil, common.CodeArgumentError, fmt.Errorf("box client_id and client_secret are required")
	}
	if redirectURI == "" {
		redirectURI = defaultBoxWebOAuthRedirectURI()
	}

	flowID := common.GenerateUUID()
	authorizationURL, err := buildBoxAuthorizationURL(clientID, redirectURI, flowID)
	if err != nil {
		return nil, common.CodeServerError, err
	}

	redisClient := connectorRedisGet()
	if redisClient == nil {
		return nil, common.CodeServerError, fmt.Errorf("not connected Redis client on the server")
	}

	state := boxWebOAuthState{
		UserID:       userID,
		AuthURL:      authorizationURL,
		ClientID:     clientID,
		ClientSecret: clientSecret,
		RedirectURI:  redirectURI,
		CreatedAt:    time.Now().Unix(),
	}
	if ok := redisClient.SetObj(ctx, webStateCacheKey(flowID, "box"), state, webFlowTTL); !ok {
		return nil, common.CodeServerError, fmt.Errorf("failed to initialize Box OAuth flow. Please verify the client configuration")
	}

	return &StartBoxWebOAuthResponse{
		FlowID:           flowID,
		AuthorizationURL: authorizationURL,
		ExpiresIn:        int64(webFlowTTL.Seconds()),
	}, common.CodeSuccess, nil
}

func (s *ConnectorService) BoxWebOAuthCallback(ctx context.Context, flowID string, oauthError string, errorDescription string, code string) string {
	flowID = strings.TrimSpace(flowID)
	if flowID == "" {
		return renderWebOAuthPopup("", false, "Missing OAuth parameters.", "box")
	}

	redisClient := connectorRedisGet()
	if redisClient == nil {
		return renderWebOAuthPopup(flowID, false, "Box OAuth session expired or invalid.", "box")
	}

	stateKey := webStateCacheKey(flowID, "box")
	var state boxWebOAuthState
	if ok := redisClient.GetObj(ctx, stateKey, &state); !ok {
		return renderWebOAuthPopup(flowID, false, "Box OAuth session expired or invalid.", "box")
	}

	if strings.TrimSpace(oauthError) != "" {
		redisClient.Delete(ctx, stateKey)
		message := strings.TrimSpace(errorDescription)
		if message == "" {
			message = strings.TrimSpace(oauthError)
		}
		if message == "" {
			message = "Authorization failed."
		}
		return renderWebOAuthPopup(flowID, false, message, "box")
	}

	code = strings.TrimSpace(code)
	if code == "" {
		return renderWebOAuthPopup(flowID, false, "Missing authorization code from Box.", "box")
	}

	token, err := exchangeBoxAuthorizationCode(state.ClientID, state.ClientSecret, state.RedirectURI, code)
	if err != nil {
		redisClient.Delete(ctx, stateKey)
		return renderWebOAuthPopup(flowID, false, "Failed to exchange tokens with Box. Please retry.", "box")
	}

	result := boxWebOAuthCredentials{
		UserID:       state.UserID,
		ClientID:     state.ClientID,
		ClientSecret: state.ClientSecret,
		AccessToken:  token.AccessToken,
		RefreshToken: token.RefreshToken,
	}
	if ok := redisClient.SetObj(ctx, webResultCacheKey(flowID, "box"), result, webFlowTTL); !ok {
		redisClient.Delete(ctx, stateKey)
		return renderWebOAuthPopup(flowID, false, "Failed to exchange tokens with Box. Please retry.", "box")
	}
	redisClient.Delete(ctx, stateKey)

	return renderWebOAuthPopup(flowID, true, "Authorization completed successfully.", "box")
}

func (s *ConnectorService) PollBoxWebOAuthResult(ctx context.Context, req *PollBoxWebOAuthResultRequest) (*PollBoxWebOAuthResultResponse, common.ErrorCode, error) {
	userID := strings.TrimSpace(permission.UserIDFromContext(ctx))
	if userID == "" {
		return nil, common.CodeUnauthorized, permission.ErrUnauthenticated
	}
	if req == nil || strings.TrimSpace(req.FlowID) == "" {
		return nil, common.CodeArgumentError, fmt.Errorf("required argument is missing: flow_id")
	}

	redisClient := connectorRedisGet()
	if redisClient == nil {
		return nil, common.CodeRunning, fmt.Errorf("authorization is still pending")
	}

	resultKey := webResultCacheKey(strings.TrimSpace(req.FlowID), "box")
	var result boxWebOAuthCredentials
	if ok := redisClient.GetObj(ctx, resultKey, &result); !ok {
		return nil, common.CodeRunning, fmt.Errorf("authorization is still pending")
	}

	if result.UserID != userID {
		return nil, common.CodePermissionError, fmt.Errorf("you are not allowed to access this authorization result")
	}

	redisClient.Delete(ctx, resultKey)
	result.UserID = ""
	return &PollBoxWebOAuthResultResponse{Credentials: result}, common.CodeSuccess, nil
}

func defaultBoxWebOAuthRedirectURI() string {
	return getEnvDefault(
		common.EnvBoxWebOAuthRedirectURI,
		"http://localhost:9380/api/v1/connectors/box/oauth/web/callback",
	)
}

func buildBoxAuthorizationURL(clientID string, redirectURI string, state string) (string, error) {
	parsedURL, err := url.Parse(boxOAuthAuthorizeURL)
	if err != nil {
		return "", err
	}

	query := parsedURL.Query()
	query.Set("client_id", clientID)
	query.Set("redirect_uri", redirectURI)
	query.Set("response_type", "code")
	query.Set("state", state)
	parsedURL.RawQuery = query.Encode()
	return parsedURL.String(), nil
}

func exchangeBoxAuthorizationCode(clientID string, clientSecret string, redirectURI string, code string) (*boxOAuthTokenResponse, error) {
	_ = redirectURI

	form := url.Values{}
	form.Set("grant_type", "authorization_code")
	form.Set("code", code)
	form.Set("client_id", clientID)
	form.Set("client_secret", clientSecret)

	ctx, cancel := context.WithTimeout(context.Background(), googleOAuthHTTPTimeout)
	defer cancel()

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, boxOAuthTokenURL, strings.NewReader(form.Encode()))
	if err != nil {
		return nil, err
	}
	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
	req.Header.Set("Accept", "application/json")

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
	if err != nil {
		return nil, err
	}

	var token boxOAuthTokenResponse
	if err = json.Unmarshal(body, &token); err != nil {
		return nil, err
	}
	if resp.StatusCode >= http.StatusBadRequest || token.Error != "" {
		if token.ErrorDesc != "" {
			return nil, errors.New(token.ErrorDesc)
		}
		if token.Error != "" {
			return nil, errors.New(token.Error)
		}
		return nil, fmt.Errorf("box token exchange failed: HTTP %d", resp.StatusCode)
	}
	if token.AccessToken == "" {
		return nil, fmt.Errorf("box token exchange failed: empty access_token")
	}
	return &token, nil
}