111 lines
3.1 KiB
Go
111 lines
3.1 KiB
Go
package processing
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
// BackfillCompanyProcessedAttributes rewrites attributes and processed_attributes
|
|
// for every processed row in a company using AttrsForPersist + category_attributes
|
|
// allowlists (same rules as processOne / enhance). Returns rows updated.
|
|
func BackfillCompanyProcessedAttributes(ctx context.Context, pool *pgxpool.Pool, companyID uuid.UUID) (int64, error) {
|
|
if pool == nil {
|
|
return 0, fmt.Errorf("backfill attributes: nil pool")
|
|
}
|
|
p := &Pipeline{Pool: pool}
|
|
sets := p.loadCategoryAttributeKeySets(ctx, companyID, uuid.Nil)
|
|
|
|
rows, err := pool.Query(ctx, `
|
|
SELECT id, COALESCE(category, ''), COALESCE(attributes, '{}'::jsonb), COALESCE(processed_attributes, '{}'::jsonb)
|
|
FROM processed_products
|
|
WHERE company_id = $1`, companyID)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("backfill attributes query: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
var updated int64
|
|
for rows.Next() {
|
|
var (
|
|
id uuid.UUID
|
|
category string
|
|
attrsJSON []byte
|
|
procAttrsJSON []byte
|
|
)
|
|
if err := rows.Scan(&id, &category, &attrsJSON, &procAttrsJSON); err != nil {
|
|
return updated, fmt.Errorf("backfill attributes scan: %w", err)
|
|
}
|
|
attrs := decodeAttrMap(attrsJSON)
|
|
procAttrs := decodeAttrMap(procAttrsJSON)
|
|
allowed := allowedAttrKeysFromSets(sets, category)
|
|
cleanAttrs := AttrsForPersist(attrs, allowed)
|
|
cleanProc := AttrsForPersist(procAttrs, allowed)
|
|
if len(cleanProc) == 0 {
|
|
cleanProc = cleanAttrs
|
|
}
|
|
newAttrsJSON, err := json.Marshal(cleanAttrs)
|
|
if err != nil {
|
|
return updated, fmt.Errorf("backfill attributes marshal attrs %s: %w", id, err)
|
|
}
|
|
newProcJSON, err := json.Marshal(cleanProc)
|
|
if err != nil {
|
|
return updated, fmt.Errorf("backfill attributes marshal processed %s: %w", id, err)
|
|
}
|
|
if bytes.Equal(compactJSON(attrsJSON), compactJSON(newAttrsJSON)) &&
|
|
bytes.Equal(compactJSON(procAttrsJSON), compactJSON(newProcJSON)) {
|
|
continue
|
|
}
|
|
ct, err := pool.Exec(ctx, `
|
|
UPDATE processed_products
|
|
SET attributes = $2::jsonb,
|
|
processed_attributes = $3::jsonb,
|
|
updated_at = now()
|
|
WHERE id = $1 AND company_id = $4`, id, string(newAttrsJSON), string(newProcJSON), companyID)
|
|
if err != nil {
|
|
return updated, fmt.Errorf("backfill attributes update %s: %w", id, err)
|
|
}
|
|
updated += ct.RowsAffected()
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return updated, fmt.Errorf("backfill attributes rows: %w", err)
|
|
}
|
|
return updated, nil
|
|
}
|
|
|
|
func decodeAttrMap(raw []byte) map[string]any {
|
|
out := map[string]any{}
|
|
if len(raw) == 0 {
|
|
return out
|
|
}
|
|
var parsed any
|
|
if err := json.Unmarshal(raw, &parsed); err != nil {
|
|
return out
|
|
}
|
|
switch t := parsed.(type) {
|
|
case map[string]any:
|
|
return t
|
|
case []any:
|
|
for _, entry := range t {
|
|
if m, ok := entry.(map[string]any); ok {
|
|
if k, ok := m["key"].(string); ok && k != "" {
|
|
out[k] = m
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func compactJSON(b []byte) []byte {
|
|
var buf bytes.Buffer
|
|
if err := json.Compact(&buf, b); err != nil {
|
|
return b
|
|
}
|
|
return buf.Bytes()
|
|
}
|