goods/api/internal/adminstore/submission.go

package adminstore

import (
	"context"
	"encoding/json"
	"errors"
	"strconv"
	"strings"
	"time"

	"github.com/jackc/pgx/v5"
)

// ErrConflict is returned when a submission has already been reviewed.
var ErrConflict = errors.New("conflict")

// SubmissionImage is one proposed image URL inside a contribution.
type SubmissionImage struct {
	URL  string `json:"url"`
	Kind string `json:"kind"`
}

// SubmissionInput is the public contribution payload (no login required).
type SubmissionInput struct {
	GTIN             *string           `json:"gtin"`
	Name             string            `json:"name"`
	BrandName        *string           `json:"brand_name"`
	CategoryID       *string           `json:"category_id"`
	NetContentValue  *float64          `json:"net_content_value"`
	NetContentUnit   *string           `json:"net_content_unit"`
	CountryOfOrigin  *string           `json:"country_of_origin"`
	IngredientsText  *string           `json:"ingredients_text"`
	Nutriments       map[string]any    `json:"nutriments"`
	Attributes       map[string]any    `json:"attributes"`
	NutritionBasis   *string           `json:"nutrition_basis"`
	ServingSize      *string           `json:"serving_size"`
	NutriScore       *string           `json:"nutri_score"`
	Images           []SubmissionImage `json:"images"`
	MSRP             []MSRPInput       `json:"msrp"`
	SubmitterName    *string           `json:"submitter_name"`
	SubmitterContact *string           `json:"submitter_contact"`
	Note             *string           `json:"note"`
	// Source tags the origin of the submission, stored inside the payload so no
	// schema change is needed. Empty means the default public contribution
	// ("community"); "backflow" marks records pushed by inventory software.
	Source *string `json:"source,omitempty"`
}

// SubmissionRow is a queue-list row for the admin review table.
type SubmissionRow struct {
	ID            string  `json:"id"`
	GTIN          *string `json:"gtin"`
	Name          string  `json:"name"`
	Status        string  `json:"status"`
	SubmitterName *string `json:"submitter_name"`
	Source        *string `json:"source"`
	Matched       bool    `json:"matched"`
	CreatedAt     string  `json:"created_at"`
	ReviewedAt    *string `json:"reviewed_at"`
}

// SubmissionDetail is the full review view of one contribution.
type SubmissionDetail struct {
	ID               string          `json:"id"`
	Status           string          `json:"status"`
	GTIN             *string         `json:"gtin"`
	Name             string          `json:"name"`
	SubmitterName    *string         `json:"submitter_name"`
	SubmitterContact *string         `json:"submitter_contact"`
	Note             *string         `json:"note"`
	ReviewNote       *string         `json:"review_note"`
	ReviewedBy       *string         `json:"reviewed_by"`
	ReviewedAt       *string         `json:"reviewed_at"`
	CreatedAt        string          `json:"created_at"`
	TargetProductID  *string         `json:"target_product_id"`
	ResultProductID  *string         `json:"result_product_id"`
	Payload          SubmissionInput `json:"payload"`
	ExistingProduct  *ProductDetail  `json:"existing_product,omitempty"`
}

// CreateSubmission validates and stores a public contribution as pending.
func (s *Store) CreateSubmission(ctx context.Context, in SubmissionInput, remoteIP string) (string, error) {
	in.Name = strings.TrimSpace(in.Name)
	if in.Name == "" {
		return "", errors.New("商品名称不能为空")
	}
	if in.GTIN != nil {
		g := strings.TrimSpace(*in.GTIN)
		if g == "" {
			in.GTIN = nil
		} else {
			in.GTIN = &g
		}
	}

	// Link to an existing product when the barcode already exists (supplement).
	var target *string
	if in.GTIN != nil {
		var pid string
		err := s.pool.QueryRow(ctx, "SELECT id FROM product WHERE gtin = $1", *in.GTIN).Scan(&pid)
		if err == nil {
			target = &pid
		} else if !errors.Is(err, pgx.ErrNoRows) {
			return "", err
		}
	}

	payload, err := json.Marshal(in)
	if err != nil {
		return "", err
	}

	var id string
	err = s.pool.QueryRow(ctx, `
INSERT INTO submission (gtin, name, payload, target_product_id, submitter_name, submitter_contact, note, remote_ip)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8) RETURNING id`,
		in.GTIN, in.Name, payload, target, in.SubmitterName, in.SubmitterContact, in.Note, remoteIP).Scan(&id)
	return id, err
}

// ListSubmissions returns submissions filtered by status (empty = all).
func (s *Store) ListSubmissions(ctx context.Context, status string, limit, offset int) ([]SubmissionRow, int, error) {
	args := []any{}
	where := "WHERE 1=1"
	if status != "" {
		args = append(args, status)
		where += " AND status = $1"
	}

	var total int
	if err := s.pool.QueryRow(ctx, "SELECT count(*) FROM submission "+where, args...).Scan(&total); err != nil {
		return nil, 0, err
	}

	args = append(args, limit, offset)
	sql := `
SELECT id, gtin, name, status, submitter_name, NULLIF(payload->>'source',''),
       (target_product_id IS NOT NULL), created_at, reviewed_at
FROM submission ` + where +
		" ORDER BY (status='pending') DESC, created_at DESC LIMIT $" +
		strconv.Itoa(len(args)-1) + " OFFSET $" + strconv.Itoa(len(args))

	rows, err := s.pool.Query(ctx, sql, args...)
	if err != nil {
		return nil, 0, err
	}
	defer rows.Close()

	out := []SubmissionRow{}
	for rows.Next() {
		var r SubmissionRow
		var created time.Time
		var reviewed *time.Time
		if err := rows.Scan(&r.ID, &r.GTIN, &r.Name, &r.Status, &r.SubmitterName, &r.Source, &r.Matched, &created, &reviewed); err != nil {
			return nil, 0, err
		}
		r.CreatedAt = created.Format(time.RFC3339)
		if reviewed != nil {
			t := reviewed.Format(time.RFC3339)
			r.ReviewedAt = &t
		}
		out = append(out, r)
	}
	return out, total, rows.Err()
}

// PendingSubmissionCount returns the number of submissions awaiting review.
func (s *Store) PendingSubmissionCount(ctx context.Context) (int, error) {
	var n int
	err := s.pool.QueryRow(ctx, "SELECT count(*) FROM submission WHERE status='pending'").Scan(&n)
	return n, err
}

// GetSubmission returns the full review detail for one submission.
func (s *Store) GetSubmission(ctx context.Context, id string) (*SubmissionDetail, error) {
	var d SubmissionDetail
	var payload []byte
	var created time.Time
	var reviewed *time.Time
	err := s.pool.QueryRow(ctx, `
SELECT id, status, gtin, name, submitter_name, submitter_contact, note,
       review_note, reviewed_by, reviewed_at, created_at, target_product_id, result_product_id, payload
FROM submission WHERE id = $1`, id).Scan(
		&d.ID, &d.Status, &d.GTIN, &d.Name, &d.SubmitterName, &d.SubmitterContact, &d.Note,
		&d.ReviewNote, &d.ReviewedBy, &reviewed, &created, &d.TargetProductID, &d.ResultProductID, &payload,
	)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, ErrNotFound
	}
	if err != nil {
		return nil, err
	}
	d.CreatedAt = created.Format(time.RFC3339)
	if reviewed != nil {
		t := reviewed.Format(time.RFC3339)
		d.ReviewedAt = &t
	}
	if len(payload) > 0 {
		_ = json.Unmarshal(payload, &d.Payload)
	}
	if d.TargetProductID != nil {
		if ep, err := s.GetProduct(ctx, *d.TargetProductID); err == nil {
			d.ExistingProduct = ep
		}
	}
	return &d, nil
}

// RejectSubmission marks a pending submission as rejected with a reviewer note.
func (s *Store) RejectSubmission(ctx context.Context, id, actor, note string) error {
	ct, err := s.pool.Exec(ctx, `
UPDATE submission SET status='rejected', review_note=$2, reviewed_by=$3, reviewed_at=now()
WHERE id=$1 AND status='pending'`, id, note, actor)
	if err != nil {
		return err
	}
	if ct.RowsAffected() == 0 {
		// Distinguish missing vs already-reviewed.
		var st string
		if e := s.pool.QueryRow(ctx, "SELECT status FROM submission WHERE id=$1", id).Scan(&st); errors.Is(e, pgx.ErrNoRows) {
			return ErrNotFound
		}
		return ErrConflict
	}
	_ = s.writeAudit(ctx, actor, "reject_submission", "submission", &id, []string{}, nil, map[string]string{"review_note": note})
	return nil
}

// ApproveSubmission applies a pending contribution to the product store
// (creating or supplementing a product), records community provenance + audit,
// recomputes quality, and marks the submission approved.
func (s *Store) ApproveSubmission(ctx context.Context, id, actor string) (*ProductDetail, error) {
	sub, err := s.GetSubmission(ctx, id)
	if err != nil {
		return nil, err
	}
	if sub.Status != "pending" {
		return nil, ErrConflict
	}
	in := sub.Payload

	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)

	communityID, err := s.sourceIDTx(ctx, tx, "community")
	if err != nil {
		return nil, err
	}

	// Resolve the target product (existing supplement vs new create).
	productID := ""
	if sub.TargetProductID != nil {
		productID = *sub.TargetProductID
	} else if in.GTIN != nil {
		var pid string
		if e := tx.QueryRow(ctx, "SELECT id FROM product WHERE gtin=$1", *in.GTIN).Scan(&pid); e == nil {
			productID = pid
		} else if !errors.Is(e, pgx.ErrNoRows) {
			return nil, e
		}
	}

	var brandID *string
	if in.BrandName != nil && strings.TrimSpace(*in.BrandName) != "" {
		bid, err := s.ensureBrand(ctx, tx, strings.TrimSpace(*in.BrandName))
		if err != nil {
			return nil, err
		}
		brandID = &bid
	}
	var gpc *string
	if in.CategoryID != nil && *in.CategoryID != "" {
		if err := tx.QueryRow(ctx, "SELECT gpc_brick_code FROM category WHERE id=$1", *in.CategoryID).Scan(&gpc); err != nil && !errors.Is(err, pgx.ErrNoRows) {
			return nil, err
		}
	}
	canonical, err := s.netCanonical(ctx, tx, in.NetContentValue, in.NetContentUnit)
	if err != nil {
		return nil, err
	}

	fields := submissionFields(in)

	var attrJSON []byte
	if len(in.Attributes) > 0 {
		attrJSON, _ = json.Marshal(in.Attributes)
	}

	if productID == "" {
		// Create a new product from the contribution.
		err = tx.QueryRow(ctx, `
INSERT INTO product (gtin, name, brand_id, category_id, gpc_brick_code,
   net_content_value, net_content_unit, net_content_canonical, country_of_origin, attributes, status)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,COALESCE($10::jsonb,'{}'::jsonb),'active') RETURNING id`,
			in.GTIN, in.Name, brandID, in.CategoryID, gpc,
			in.NetContentValue, in.NetContentUnit, canonical, in.CountryOfOrigin, attrJSON).Scan(&productID)
		if err != nil {
			return nil, err
		}
	} else {
		// Supplement an existing product: only overwrite fields the
		// contribution actually provides (COALESCE keeps current values).
		_, err = tx.Exec(ctx, `
UPDATE product SET
   name=COALESCE(NULLIF($2,''), name),
   brand_id=COALESCE($3, brand_id),
   category_id=COALESCE($4, category_id),
   gpc_brick_code=COALESCE($5, gpc_brick_code),
   net_content_value=COALESCE($6, net_content_value),
   net_content_unit=COALESCE($7, net_content_unit),
   net_content_canonical=COALESCE($8, net_content_canonical),
   country_of_origin=COALESCE($9, country_of_origin),
   gtin=COALESCE($10, gtin),
   attributes=product.attributes || COALESCE($11::jsonb,'{}'::jsonb)
WHERE id=$1`,
			productID, in.Name, brandID, in.CategoryID, gpc,
			in.NetContentValue, in.NetContentUnit, canonical, in.CountryOfOrigin, in.GTIN, attrJSON)
		if err != nil {
			return nil, err
		}
	}

	// food_detail: upsert only when the contribution carries food data, so a
	// non-food submission (drug/3C/generic) doesn't create an empty row.
	var nutriJSON []byte
	if len(in.Nutriments) > 0 {
		nutriJSON, _ = json.Marshal(in.Nutriments)
	}
	foodPresent := len(in.Nutriments) > 0 ||
		(in.IngredientsText != nil && *in.IngredientsText != "") ||
		(in.NutritionBasis != nil && *in.NutritionBasis != "") ||
		(in.ServingSize != nil && *in.ServingSize != "") ||
		(in.NutriScore != nil && *in.NutriScore != "")
	if foodPresent {
		_, err = tx.Exec(ctx, `
INSERT INTO food_detail (product_id, ingredients_text, nutriments, nutrition_basis, serving_size, nutri_score)
VALUES ($1,$2,$3,$4,$5,$6)
ON CONFLICT (product_id) DO UPDATE SET
   ingredients_text=COALESCE(EXCLUDED.ingredients_text, food_detail.ingredients_text),
   nutriments=COALESCE(EXCLUDED.nutriments, food_detail.nutriments),
   nutrition_basis=COALESCE(EXCLUDED.nutrition_basis, food_detail.nutrition_basis),
   serving_size=COALESCE(EXCLUDED.serving_size, food_detail.serving_size),
   nutri_score=COALESCE(EXCLUDED.nutri_score, food_detail.nutri_score)`,
			productID, in.IngredientsText, nutriJSON, in.NutritionBasis, in.ServingSize, in.NutriScore)
		if err != nil {
			return nil, err
		}
	}

	for _, im := range in.Images {
		url := strings.TrimSpace(im.URL)
		if url == "" {
			continue
		}
		kind := im.Kind
		if kind != "front" && kind != "ingredients" && kind != "nutrition" {
			kind = "other"
		}
		if _, err := tx.Exec(ctx, `
INSERT INTO product_image (product_id, url, kind, source_id) VALUES ($1,$2,$3,$4)`,
			productID, url, kind, communityID); err != nil {
			return nil, err
		}
	}

	for _, m := range in.MSRP {
		if m.Amount <= 0 {
			continue
		}
		cur := m.Currency
		if cur == "" {
			cur = "CNY"
		}
		region := m.Region
		if region == "" {
			region = "CN"
		}
		if _, err := tx.Exec(ctx, `
INSERT INTO product_msrp (product_id, amount, currency, region, source_id, source_url, effective_date, note)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8)`,
			productID, m.Amount, cur, region, communityID, m.SourceURL, m.EffectiveDate, m.Note); err != nil {
			return nil, err
		}
	}

	if _, err := s.recomputeQualityTx(ctx, tx, productID); err != nil {
		return nil, err
	}

	if _, err := tx.Exec(ctx, `
UPDATE submission SET status='approved', reviewed_by=$2, reviewed_at=now(), result_product_id=$3
WHERE id=$1`, id, actor, productID); err != nil {
		return nil, err
	}

	// Field-level provenance for the contributed fields (community source).
	if len(fields) > 0 {
		if _, err := tx.Exec(ctx, `
INSERT INTO product_source (product_id, source_id, url, fields, fetched_at, raw)
VALUES ($1,$2,NULL,$3,now(),NULL)`, productID, communityID, fields); err != nil {
			return nil, err
		}
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}

	_ = s.writeAudit(ctx, actor, "approve_submission", "product", &productID, fields,
		map[string]string{"submission_id": id}, map[string]string{"product_id": productID})

	return s.GetProduct(ctx, productID)
}

func (s *Store) sourceIDTx(ctx context.Context, tx pgx.Tx, name string) (string, error) {
	var id string
	err := tx.QueryRow(ctx, "SELECT id FROM source WHERE name=$1", name).Scan(&id)
	return id, err
}

// submissionFields lists the product fields a contribution provides values for.
func submissionFields(in SubmissionInput) []string {
	fields := []string{"name"}
	add := func(name string, present bool) {
		if present {
			fields = append(fields, name)
		}
	}
	add("gtin", in.GTIN != nil && *in.GTIN != "")
	add("brand", in.BrandName != nil && strings.TrimSpace(*in.BrandName) != "")
	add("category", in.CategoryID != nil && *in.CategoryID != "")
	add("net_content", in.NetContentValue != nil)
	add("country_of_origin", in.CountryOfOrigin != nil && *in.CountryOfOrigin != "")
	add("ingredients", in.IngredientsText != nil && *in.IngredientsText != "")
	add("nutriments", len(in.Nutriments) > 0)
	add("image", len(in.Images) > 0)
	for k, v := range in.Attributes {
		if v == nil {
			continue
		}
		if s, ok := v.(string); ok && s == "" {
			continue
		}
		fields = append(fields, k)
	}
	return fields
}