diff --git a/api/internal/adminhandler/handler.go b/api/internal/adminhandler/handler.go index 475225a..6a52bfe 100644 --- a/api/internal/adminhandler/handler.go +++ b/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") diff --git a/api/internal/adminstore/backflow.go b/api/internal/adminstore/backflow.go new file mode 100644 index 0000000..a6fcbcb --- /dev/null +++ b/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 +} diff --git a/api/internal/adminstore/submission.go b/api/internal/adminstore/submission.go index 3ab94c0..c504008 100644 --- a/api/internal/adminstore/submission.go +++ b/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)