feat(admin): 新增档案回流接口 POST /api/public/backflow

进销存软件可用公开 API Key(og_live_) 批量回流未收录商品,进入现有审核
队列,审核通过后收录。按 GTIN 去重(已收录跳过 exists,已有待审跳过
duplicate),来源标记 source=backflow,在后台队列与公众投稿区分。

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

api/internal/adminhandler/handler.go

@@ -14,6 +14,7 @@ import (
 	"github.com/go-chi/chi/v5/middleware"
 
 	"github.com/baicai2026-baicai/goods/api/internal/adminstore"
+	"github.com/baicai2026-baicai/goods/api/internal/apikey"
 	"github.com/baicai2026-baicai/goods/api/internal/auth"
 	"github.com/baicai2026-baicai/goods/api/internal/gtin"
 	"github.com/baicai2026-baicai/goods/api/internal/ratelimit"
@@ -110,6 +111,11 @@ func (h *Handler) Router() http.Handler {
 	// admin approves them.
 	r.Post("/api/public/submissions", h.CreateSubmission)
 
+	// Archive backflow: inventory-management software pushes products missing
+	// from the archive. Authenticated with a public API key; records enter the
+	// same moderation queue and are archived only after admin approval.
+	r.Post("/api/public/backflow", h.Backflow)
+
 	return r
 }
 
@@ -634,6 +640,62 @@ func (h *Handler) CreateSubmission(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusCreated, map[string]string{"id": id, "status": "pending"})
 }
 
+// Backflow accepts a batch of products pushed by inventory-management software.
+// It authenticates with a public API key (X-API-Key or Bearer), enqueues each
+// item for admin review (deduplicating by GTIN), and returns a per-item summary.
+func (h *Handler) Backflow(w http.ResponseWriter, r *http.Request) {
+	raw := presentedAPIKey(r)
+	if raw == "" {
+		writeError(w, http.StatusUnauthorized, "missing_api_key", "缺少 API key（请在 X-API-Key 或 Authorization: Bearer 中提供）")
+		return
+	}
+	if !apikey.IsWellFormed(raw) {
+		writeError(w, http.StatusUnauthorized, "invalid_api_key", "API key 格式无效")
+		return
+	}
+	if _, err := h.store.APIKeyByHash(r.Context(), apikey.Hash(raw)); err != nil {
+		if errors.Is(err, adminstore.ErrNotFound) {
+			writeError(w, http.StatusUnauthorized, "invalid_api_key", "API key 无效或已吊销")
+			return
+		}
+		writeError(w, http.StatusInternalServerError, "internal_error", err.Error())
+		return
+	}
+
+	var items []adminstore.SubmissionInput
+	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 16<<20)).Decode(&items); err != nil {
+		writeError(w, http.StatusBadRequest, "bad_request", "请求体应为商品数组 (JSON array)")
+		return
+	}
+
+	if len(items) == 0 {
+		writeError(w, http.StatusBadRequest, "bad_request", "回流列表为空")
+		return
+	}
+	if len(items) > 1000 {
+		writeError(w, http.StatusBadRequest, "too_many", "单次回流最多 1000 条")
+		return
+	}
+
+	sum, err := h.store.CreateBackflowSubmissions(r.Context(), items, realIP(r))
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, "internal_error", err.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, sum)
+}
+
+// presentedAPIKey extracts a public API key from X-API-Key or a Bearer token.
+func presentedAPIKey(r *http.Request) string {
+	if v := strings.TrimSpace(r.Header.Get("X-API-Key")); v != "" {
+		return v
+	}
+	if v := r.Header.Get("Authorization"); strings.HasPrefix(v, "Bearer ") {
+		return strings.TrimSpace(strings.TrimPrefix(v, "Bearer "))
+	}
+	return ""
+}
+
 // ListSubmissions returns the moderation queue (admin).
 func (h *Handler) ListSubmissions(w http.ResponseWriter, r *http.Request) {
 	status := r.URL.Query().Get("status")

api/internal/adminstore/backflow.go

@@ -0,0 +1,145 @@
+package adminstore
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"strings"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// backflowSource is the value stored in submission.payload->>'source' for
+// records pushed by inventory-management software via the backflow API.
+const backflowSource = "backflow"
+
+// APIKeyAuth is the minimal key metadata needed to authenticate a backflow
+// caller. Only active (non-revoked) keys resolve.
+type APIKeyAuth struct {
+	ID   string
+	Name string
+}
+
+// APIKeyByHash returns the active key matching a SHA-256 hash, or ErrNotFound
+// when no such active key exists. Used to authenticate machine callers (e.g.
+// the backflow endpoint) with the same public API keys issued to API users.
+func (s *Store) APIKeyByHash(ctx context.Context, hash string) (*APIKeyAuth, error) {
+	var k APIKeyAuth
+	err := s.pool.QueryRow(ctx,
+		"SELECT id, name FROM api_key WHERE key_hash = $1 AND revoked_at IS NULL", hash,
+	).Scan(&k.ID, &k.Name)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &k, nil
+}
+
+// BackflowResult reports the outcome of one item in a backflow batch.
+type BackflowResult struct {
+	GTIN   *string `json:"gtin"`
+	Name   string  `json:"name"`
+	Status string  `json:"status"`       // queued | exists | duplicate | invalid
+	ID     string  `json:"id,omitempty"` // submission id when queued
+	Reason string  `json:"reason,omitempty"`
+}
+
+// BackflowSummary aggregates a backflow batch outcome.
+type BackflowSummary struct {
+	Total     int              `json:"total"`
+	Queued    int              `json:"queued"`
+	Exists    int              `json:"exists"`
+	Duplicate int              `json:"duplicate"`
+	Invalid   int              `json:"invalid"`
+	Results   []BackflowResult `json:"results"`
+}
+
+// CreateBackflowSubmissions enqueues products pushed by inventory software for
+// admin review. Each item is deduplicated by GTIN: items whose barcode already
+// matches an archived product are skipped ("exists"), and items that duplicate
+// a pending submission are skipped ("duplicate"). Accepted items are tagged
+// with source="backflow" and enter the same moderation queue as public
+// contributions; approval creates the product exactly as ApproveSubmission does.
+//
+// Items are processed independently: a bad item never rolls back accepted ones.
+func (s *Store) CreateBackflowSubmissions(ctx context.Context, items []SubmissionInput, remoteIP string) (BackflowSummary, error) {
+	sum := BackflowSummary{Total: len(items), Results: make([]BackflowResult, 0, len(items))}
+
+	for _, in := range items {
+		in.Name = strings.TrimSpace(in.Name)
+		res := BackflowResult{Name: in.Name}
+
+		if in.GTIN != nil {
+			g := strings.TrimSpace(*in.GTIN)
+			if g == "" {
+				in.GTIN = nil
+			} else {
+				in.GTIN = &g
+				res.GTIN = &g
+			}
+		}
+
+		if in.Name == "" {
+			res.Status = "invalid"
+			res.Reason = "商品名称不能为空"
+			sum.Invalid++
+			sum.Results = append(sum.Results, res)
+			continue
+		}
+
+		if in.GTIN != nil {
+			// Already archived: backflow only carries products we don't have yet.
+			var pid string
+			err := s.pool.QueryRow(ctx, "SELECT id FROM product WHERE gtin = $1", *in.GTIN).Scan(&pid)
+			if err == nil {
+				res.Status = "exists"
+				res.Reason = "该条码商品已收录"
+				sum.Exists++
+				sum.Results = append(sum.Results, res)
+				continue
+			} else if !errors.Is(err, pgx.ErrNoRows) {
+				return sum, err
+			}
+
+			// Collapse repeated auto-pushes of the same barcode in the queue.
+			var sid string
+			err = s.pool.QueryRow(ctx,
+				"SELECT id FROM submission WHERE gtin = $1 AND status = 'pending' LIMIT 1", *in.GTIN).Scan(&sid)
+			if err == nil {
+				res.Status = "duplicate"
+				res.Reason = "已有待审核的同条码回流记录"
+				res.ID = sid
+				sum.Duplicate++
+				sum.Results = append(sum.Results, res)
+				continue
+			} else if !errors.Is(err, pgx.ErrNoRows) {
+				return sum, err
+			}
+		}
+
+		src := backflowSource
+		in.Source = &src
+
+		payload, err := json.Marshal(in)
+		if err != nil {
+			return sum, err
+		}
+
+		var id string
+		err = s.pool.QueryRow(ctx, `
+INSERT INTO submission (gtin, name, payload, submitter_name, submitter_contact, note, remote_ip)
+VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING id`,
+			in.GTIN, in.Name, payload, in.SubmitterName, in.SubmitterContact, in.Note, remoteIP).Scan(&id)
+		if err != nil {
+			return sum, err
+		}
+		res.Status = "queued"
+		res.ID = id
+		sum.Queued++
+		sum.Results = append(sum.Results, res)
+	}
+
+	return sum, nil
+}

api/internal/adminstore/submission.go

@@ -40,6 +40,10 @@ type SubmissionInput struct {
 	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.
@@ -49,6 +53,7 @@ type SubmissionRow struct {
 	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"`
@@ -129,8 +134,8 @@ func (s *Store) ListSubmissions(ctx context.Context, status string, limit, offse
 
 	args = append(args, limit, offset)
 	sql := `
-SELECT id, gtin, name, status, submitter_name, (target_product_id IS NOT NULL),
-       created_at, reviewed_at
+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))
@@ -146,7 +151,7 @@ FROM submission ` + where +
 		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.Matched, &created, &reviewed); err != nil {
+		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)