greeneclipse
8580c996c3
Drop one-shot tmp/axe scripts and agent i18n scratch so the Gitea tree is deployable.
498 lines
16 KiB
Go
498 lines
16 KiB
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()
|
|
}
|