goods/api/internal/adminstore/backflow.go

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
}