descrybe/apps/api/internal/shopify/service.go

package shopify

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"strings"
	"time"

	"github.com/descrybe/descrybe-v2/apps/api/internal/security"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

var (
	ErrNotConfigured           = errors.New("shopify not configured")
	ErrNotEnabled              = errors.New("shopify sync is disabled")
	ErrMissingCreds            = errors.New("shopify credentials missing")
	ErrInvalidSyncScope        = errors.New("invalid product sync scope")
	ErrInvalidScheduleInterval = errors.New("schedule_interval_hours must be between 0 and 168")
)

type Service struct {
	Pool       *pgxpool.Pool
	Key        []byte
	HTTPClient *http.Client
}

type Config struct {
	ShopDomain            string       `json:"shop_domain"`
	APIVersion            string       `json:"api_version"`
	IsEnabled             bool         `json:"is_enabled"`
	Configured            bool         `json:"configured"`
	LastSyncedAt          *time.Time   `json:"last_synced_at,omitempty"`
	LastTestAt            *time.Time   `json:"last_test_at,omitempty"`
	LastTestStatus        *string      `json:"last_test_status,omitempty"`
	HasCredentials        bool         `json:"has_credentials"`
	HasClientCredentials  bool         `json:"has_client_credentials"`
	AuthMode              string       `json:"auth_mode,omitempty"`
	PendingSync           bool         `json:"pending_sync"`
	PendingOrdersSync     bool         `json:"pending_orders_sync"`
	MatchStrategy         string       `json:"match_strategy"`
	LastSyncStatus        string       `json:"last_sync_status,omitempty"`
	LastSyncError         string       `json:"last_sync_error,omitempty"`
	LastSyncSummary       *SyncSummary `json:"last_sync_summary,omitempty"`
	ProductMapCount       int          `json:"product_map_count"`
	SyncLimit             int          `json:"sync_limit,omitempty"`
	LastOrdersSyncedAt    *time.Time   `json:"last_orders_synced_at,omitempty"`
	LastOrdersSyncStatus  string       `json:"last_orders_sync_status,omitempty"`
	LastOrdersSyncError   string       `json:"last_orders_sync_error,omitempty"`
	DryRun                bool         `json:"dry_run"`
	ReviewsSupported      bool         `json:"reviews_supported"`
	ScheduleIntervalHours int          `json:"schedule_interval_hours"`
	SchedulePaused        bool         `json:"schedule_paused"`
}

// UpdateInput is the PUT /api/shopify body (secrets never echoed back).
type UpdateInput struct {
	ShopDomain   string
	AccessToken  string
	APIVersion   string
	ClientID     string
	ClientSecret string
	IsEnabled    bool
	DryRun       bool
}

func NewService(pool *pgxpool.Pool, key []byte) *Service {
	return &Service{
		Pool: pool,
		Key:  key,
		// Dial-time SSRF; no loopback (Admin API is always public *.myshopify.com).
		HTTPClient: security.SafeHTTPClient(defaultTimeout, false),
	}
}

type storedConfig struct {
	shopDomain, tokenEnc, apiVersion string
	enabled                          bool
	syncOptions                      []byte
	lastSync, lastTest               *time.Time
	status                           *string
}

func (s *Service) loadStored(ctx context.Context, companyID uuid.UUID) (storedConfig, error) {
	var sc storedConfig
	err := s.Pool.QueryRow(ctx, `
		SELECT shop_domain, access_token, api_version, is_enabled, sync_options, last_synced_at, last_test_at, last_test_status
		FROM shopify_configs WHERE company_id = $1`, companyID).Scan(
		&sc.shopDomain, &sc.tokenEnc, &sc.apiVersion, &sc.enabled, &sc.syncOptions, &sc.lastSync, &sc.lastTest, &sc.status,
	)
	return sc, err
}

func (s *Service) GetConfig(ctx context.Context, companyID uuid.UUID) (Config, error) {
	sc, err := s.loadStored(ctx, companyID)
	if errors.Is(err, pgx.ErrNoRows) {
		return Config{}, err
	}
	if err != nil {
		return Config{}, err
	}
	opt := parseSyncOptions(sc.syncOptions)
	token, _ := DecryptSecret(s.Key, sc.tokenEnc)
	apiVersion := normalizeAPIVersion(sc.apiVersion)
	hasClientCreds := opt.ClientID != "" && opt.ClientSecretEnc != ""
	authMode := strings.TrimSpace(opt.AuthMode)
	if authMode == "" && hasClientCreds {
		authMode = authModeClientCredentials
	}
	if authMode == "" && token != "" {
		authMode = authModeLegacyToken
	}
	return Config{
		ShopDomain:            sc.shopDomain,
		APIVersion:            apiVersion,
		IsEnabled:             sc.enabled,
		Configured:            true,
		LastSyncedAt:          sc.lastSync,
		LastTestAt:            sc.lastTest,
		LastTestStatus:        sc.status,
		HasCredentials:        token != "" || hasClientCreds,
		HasClientCredentials:  hasClientCreds,
		AuthMode:              authMode,
		PendingSync:           opt.PendingSync,
		PendingOrdersSync:     opt.PendingOrdersSync,
		MatchStrategy:         opt.MatchStrategy,
		LastSyncStatus:        opt.LastSyncStatus,
		LastSyncError:         opt.LastSyncError,
		LastSyncSummary:       opt.LastSyncSummary,
		ProductMapCount:       len(opt.ProductIDs),
		LastOrdersSyncedAt:    opt.LastOrdersSyncedAt,
		LastOrdersSyncStatus:  opt.LastOrdersSyncStatus,
		LastOrdersSyncError:   opt.LastOrdersSyncError,
		DryRun:                opt.DryRun || strings.EqualFold(token, dryRunToken),
		ReviewsSupported:      false,
		ScheduleIntervalHours: opt.ScheduleIntervalHours,
		SchedulePaused:        opt.SchedulePaused,
	}, nil
}

func (s *Service) UpdateConfig(ctx context.Context, companyID uuid.UUID, in UpdateInput) (Config, error) {
	normalized, err := NormalizeShopDomain(in.ShopDomain)
	if err != nil {
		return Config{}, err
	}
	apiVersion := normalizeAPIVersion(in.APIVersion)
	clientID := strings.TrimSpace(in.ClientID)
	clientSecret := strings.TrimSpace(in.ClientSecret)
	accessToken := strings.TrimSpace(in.AccessToken)

	if (clientID == "") != (clientSecret == "") {
		return Config{}, ErrInvalidClientCredentials
	}

	sc, loadErr := s.loadStored(ctx, companyID)
	opt := SyncOptions{}
	if loadErr == nil {
		opt = parseSyncOptions(sc.syncOptions)
	} else if !errors.Is(loadErr, pgx.ErrNoRows) {
		return Config{}, loadErr
	}
	opt.DryRun = in.DryRun

	tokenEnc := ""
	switch {
	case clientID != "" && clientSecret != "":
		secretEnc, encErr := EncryptSecret(s.Key, clientSecret)
		if encErr != nil {
			return Config{}, encErr
		}
		opt.AuthMode = authModeClientCredentials
		opt.ClientID = clientID
		opt.ClientSecretEnc = secretEnc
		if in.DryRun || strings.EqualFold(accessToken, dryRunToken) {
			tokenEnc, err = EncryptSecret(s.Key, dryRunToken)
			if err != nil {
				return Config{}, err
			}
			opt.TokenExpiresAt = nil
		} else {
			tok, exErr := ExchangeClientCredentials(ctx, s.HTTPClient, normalized, clientID, clientSecret)
			if exErr != nil {
				return Config{}, exErr
			}
			tokenEnc, err = EncryptSecret(s.Key, tok.AccessToken)
			if err != nil {
				return Config{}, err
			}
			exp := tok.ExpiresAt
			opt.TokenExpiresAt = &exp
		}
	case accessToken != "":
		tokenEnc, err = EncryptSecret(s.Key, accessToken)
		if err != nil {
			return Config{}, err
		}
		opt.AuthMode = authModeLegacyToken
		opt.ClientID = ""
		opt.ClientSecretEnc = ""
		opt.TokenExpiresAt = nil
	}

	rawOpt, err := json.Marshal(opt)
	if err != nil {
		return Config{}, err
	}
	_, err = s.Pool.Exec(ctx, `
		INSERT INTO shopify_configs (company_id, shop_domain, access_token, api_version, is_enabled, sync_options, updated_at)
		VALUES ($1, $2, $3, $4, $5, $6, now())
		ON CONFLICT (company_id) DO UPDATE SET
			shop_domain = EXCLUDED.shop_domain,
			access_token = CASE WHEN EXCLUDED.access_token <> '' THEN EXCLUDED.access_token ELSE shopify_configs.access_token END,
			api_version = EXCLUDED.api_version,
			is_enabled = EXCLUDED.is_enabled,
			sync_options = EXCLUDED.sync_options,
			updated_at = now()`,
		companyID, normalized, tokenEnc, apiVersion, in.IsEnabled, rawOpt)
	if err != nil {
		return Config{}, err
	}
	return s.GetConfig(ctx, companyID)
}

// UpdateSchedule sets auto product-sync interval hours and pause flag on sync_options.
// hours 0 means the worker default (6h). Manual sync remains available when paused.
func (s *Service) UpdateSchedule(ctx context.Context, companyID uuid.UUID, hours int, paused bool) (Config, error) {
	if hours < 0 || hours > maxScheduleIntervalH {
		return Config{}, ErrInvalidScheduleInterval
	}
	sc, err := s.loadStored(ctx, companyID)
	if errors.Is(err, pgx.ErrNoRows) {
		return Config{}, ErrNotConfigured
	}
	if err != nil {
		return Config{}, err
	}
	opt := parseSyncOptions(sc.syncOptions)
	opt.ScheduleIntervalHours = hours
	opt.SchedulePaused = paused
	raw, err := json.Marshal(opt)
	if err != nil {
		return Config{}, err
	}
	_, err = s.Pool.Exec(ctx, `
		UPDATE shopify_configs SET sync_options = $2, updated_at = now() WHERE company_id = $1`,
		companyID, raw)
	if err != nil {
		return Config{}, err
	}
	return s.GetConfig(ctx, companyID)
}

func (s *Service) clientFor(ctx context.Context, companyID uuid.UUID) (*Client, storedConfig, SyncOptions, error) {
	sc, err := s.loadStored(ctx, companyID)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, sc, SyncOptions{}, ErrNotConfigured
	}
	if err != nil {
		return nil, sc, SyncOptions{}, err
	}
	opt := parseSyncOptions(sc.syncOptions)
	token, err := DecryptSecret(s.Key, sc.tokenEnc)
	if err != nil {
		return nil, sc, SyncOptions{}, err
	}
	if sc.shopDomain == "" {
		return nil, sc, SyncOptions{}, ErrMissingCreds
	}

	hasClientCreds := opt.ClientID != "" && opt.ClientSecretEnc != ""
	if token == "" && !hasClientCreds {
		return nil, sc, SyncOptions{}, ErrMissingCreds
	}

	if !opt.DryRun && hasClientCreds && (token == "" || strings.EqualFold(token, dryRunToken) || tokenNeedsRefresh(opt.TokenExpiresAt, time.Now().UTC())) {
		secret, decErr := DecryptSecret(s.Key, opt.ClientSecretEnc)
		if decErr != nil {
			return nil, sc, SyncOptions{}, decErr
		}
		tok, exErr := ExchangeClientCredentials(ctx, s.HTTPClient, sc.shopDomain, opt.ClientID, secret)
		if exErr != nil {
			return nil, sc, SyncOptions{}, exErr
		}
		tokenEnc, encErr := EncryptSecret(s.Key, tok.AccessToken)
		if encErr != nil {
			return nil, sc, SyncOptions{}, encErr
		}
		exp := tok.ExpiresAt
		opt.AuthMode = authModeClientCredentials
		opt.TokenExpiresAt = &exp
		rawOpt, marshalErr := json.Marshal(opt)
		if marshalErr != nil {
			return nil, sc, SyncOptions{}, marshalErr
		}
		if _, execErr := s.Pool.Exec(ctx, `
			UPDATE shopify_configs SET access_token = $2, sync_options = $3, updated_at = now()
			WHERE company_id = $1`, companyID, tokenEnc, rawOpt); execErr != nil {
			return nil, sc, SyncOptions{}, execErr
		}
		sc.tokenEnc = tokenEnc
		sc.syncOptions = rawOpt
		token = tok.AccessToken
	}

	if token == "" {
		return nil, sc, SyncOptions{}, ErrMissingCreds
	}

	client := NewClient(sc.shopDomain, token, sc.apiVersion, s.HTTPClient)
	if opt.DryRun {
		client.DryRun = true
	}
	return client, sc, opt, nil
}

func (s *Service) TestConnection(ctx context.Context, companyID uuid.UUID) (map[string]any, error) {
	client, _, _, err := s.clientFor(ctx, companyID)
	status := "ok"
	message := "connection successful"
	if err != nil {
		status = "failed"
		message = err.Error()
		_, _ = s.Pool.Exec(ctx, `
			UPDATE shopify_configs SET last_test_at = now(), last_test_status = $2, updated_at = now()
			WHERE company_id = $1`, companyID, status)
		return map[string]any{"status": status, "message": message}, err
	}
	shop, err := client.TestConnection(ctx)
	if err != nil {
		status = "failed"
		message = "connection failed"
		_, execErr := s.Pool.Exec(ctx, `
			UPDATE shopify_configs SET last_test_at = now(), last_test_status = $2, updated_at = now()
			WHERE company_id = $1`, companyID, status)
		if execErr != nil {
			return map[string]any{"status": status, "message": message}, execErr
		}
		return map[string]any{"status": status, "message": message, "dry_run": client.DryRun}, err
	}
	if _, err := s.Pool.Exec(ctx, `
		UPDATE shopify_configs SET last_test_at = now(), last_test_status = $2, updated_at = now()
		WHERE company_id = $1`, companyID, status); err != nil {
		return map[string]any{"status": status, "message": message}, err
	}
	out := map[string]any{"status": status, "message": message, "dry_run": client.DryRun}
	if shop != nil {
		out["shop_name"] = shop.Name
		out["shop_domain"] = shop.Domain
		out["currency"] = shop.Currency
	}
	return out, nil
}

func (s *Service) EnqueueSync(ctx context.Context, companyID uuid.UUID, scopes ...ProductSyncScope) (map[string]any, error) {
	if err := s.requireEnabledCreds(ctx, companyID); err != nil {
		return nil, err
	}
	sc, err := s.loadStored(ctx, companyID)
	if err != nil {
		return nil, err
	}
	opt := parseSyncOptions(sc.syncOptions)
	if len(scopes) > 0 {
		if err := applyProductSyncScope(&opt, scopes[0]); err != nil {
			return nil, err
		}
	} else {
		clearOneShotSyncFilters(&opt)
	}
	opt.PendingSync = true
	if err := s.saveSyncOptions(ctx, companyID, opt); err != nil {
		return nil, err
	}
	return map[string]any{"status": "accepted", "message": "shopify product sync queued"}, nil
}

func (s *Service) EnqueueOrdersSync(ctx context.Context, companyID uuid.UUID) (map[string]any, error) {
	if err := s.requireEnabledCreds(ctx, companyID); err != nil {
		return nil, err
	}
	sc, err := s.loadStored(ctx, companyID)
	if err != nil {
		return nil, err
	}
	opt := parseSyncOptions(sc.syncOptions)
	opt.PendingOrdersSync = true
	if err := s.saveSyncOptions(ctx, companyID, opt); err != nil {
		return nil, err
	}
	return map[string]any{"status": "accepted", "message": "shopify orders sync queued"}, nil
}

func (s *Service) requireEnabledCreds(ctx context.Context, companyID uuid.UUID) error {
	sc, err := s.loadStored(ctx, companyID)
	if errors.Is(err, pgx.ErrNoRows) {
		return ErrNotConfigured
	}
	if err != nil {
		return err
	}
	if !sc.enabled {
		return ErrNotEnabled
	}
	token, err := DecryptSecret(s.Key, sc.tokenEnc)
	if err != nil {
		return err
	}
	opt := parseSyncOptions(sc.syncOptions)
	hasClientCreds := opt.ClientID != "" && opt.ClientSecretEnc != ""
	if sc.shopDomain == "" || (token == "" && !hasClientCreds) {
		return ErrMissingCreds
	}
	return nil
}

func (s *Service) saveSyncOptions(ctx context.Context, companyID uuid.UUID, opt SyncOptions) error {
	pruneStringInt64Map(opt.ProductIDs, maxProductIDMap)
	raw, err := json.Marshal(opt)
	if err != nil {
		return err
	}
	_, err = s.Pool.Exec(ctx, `
		UPDATE shopify_configs SET sync_options = $2, updated_at = now()
		WHERE company_id = $1`, companyID, raw)
	return err
}

// ClaimNextPendingJob claims the next pending Shopify sync (products or orders).
func (s *Service) ClaimNextPendingJob(ctx context.Context) (uuid.UUID, string, error) {
	var companyID uuid.UUID
	var kind string
	err := s.Pool.QueryRow(ctx, `
		WITH candidate AS (
			SELECT company_id,
				CASE
					WHEN COALESCE(sync_options->>'pending_sync', 'false') = 'true' THEN 'products'
					WHEN COALESCE(sync_options->>'pending_orders_sync', 'false') = 'true' THEN 'orders'
					ELSE ''
				END AS kind
			FROM shopify_configs
			WHERE is_enabled = true
				AND (
					COALESCE(sync_options->>'pending_sync', 'false') = 'true'
					OR COALESCE(sync_options->>'pending_orders_sync', 'false') = 'true'
				)
			ORDER BY updated_at ASC
			LIMIT 1
			FOR UPDATE SKIP LOCKED
		)
		UPDATE shopify_configs c
		SET sync_options = CASE candidate.kind
				WHEN 'products' THEN jsonb_set(COALESCE(c.sync_options, '{}'::jsonb), '{pending_sync}', 'false'::jsonb, true)
				WHEN 'orders' THEN jsonb_set(COALESCE(c.sync_options, '{}'::jsonb), '{pending_orders_sync}', 'false'::jsonb, true)
				ELSE c.sync_options
			END,
			updated_at = now()
		FROM candidate
		WHERE c.company_id = candidate.company_id AND candidate.kind <> ''
		RETURNING c.company_id, candidate.kind`).Scan(&companyID, &kind)
	return companyID, kind, err
}

// EnqueueDueScheduled marks enabled Shopify configs pending when last_synced_at is stale.
func (s *Service) EnqueueDueScheduled(ctx context.Context, defaultInterval time.Duration) (int, error) {
	rows, err := s.Pool.Query(ctx, `
		SELECT company_id, sync_options, last_synced_at
		FROM shopify_configs
		WHERE is_enabled = true
			AND COALESCE(sync_options->>'pending_sync', 'false') <> 'true'`)
	if err != nil {
		return 0, err
	}
	defer rows.Close()

	n := 0
	now := time.Now().UTC()
	for rows.Next() {
		var companyID uuid.UUID
		var raw []byte
		var last *time.Time
		if err := rows.Scan(&companyID, &raw, &last); err != nil {
			return n, err
		}
		opt := parseSyncOptions(raw)
		if !shouldEnqueueScheduled(opt.SchedulePaused, opt.ScheduleIntervalHours, last, now, defaultInterval) {
			continue
		}
		if _, err := s.EnqueueSync(ctx, companyID); err != nil {
			continue
		}
		n++
	}
	return n, rows.Err()
}