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 }