`"britannia"` is a substring of all 258 Britannia product names, and textScore returned 0.95 for any product whose name contained the label. So every one of them tied, the tie broke alphabetically, and the customer was shown one arbitrary biscuit with "confidence": 0.95 and a price. Lens hands back a bare wordmark often — it is usually the biggest thing printed on a packet — so this was the common case, not an edge one. Found via the example request in the mobile team's own proposal. Scoring now asks both questions. A hit carries `score` (ranks) and `text` (how specifically the label names THIS product: the harmonic mean of how much of the label the product explains and how much of the product's name the label explains, pack sizes dropped from both sides). A brand name scores its products ~0.33 equally instead of 0.95 arbitrarily. The "vector and text agree" bonus is now proportional to the text score, so a weak match can no longer inflate a whole brand. isAmbiguous reads that: the leader is a guess if anything is level with it (margin) or if the label names no one product (specificity), and then the response carries `ambiguous: true` with `candidates` — distinct products, not pack sizes, at most ten, each marked with whether one of the customer's stores has it in stock, available ones first. `match` is nil and `stores` empty on that path: no price for a product nobody chose. Erring towards asking is deliberate — a tap versus the wrong biscuit. To act on a pick, /lookup now accepts `brand` + `catalogueid` instead of a label and skips recognition entirely (also serves deep links and re-order). New: ScanRepository.CatalogueRef, resolving via the brand tables discovered from information_schema, never a name built from the request. Also: scratch/cataloguedims now reports every vector column, not just `embedding` — which is how we learned the catalogue also carries img_vector(1024), filled on 1885 of 2124 rows. SCAN_TO_ORDER.md records why that column stays unread for now and what would change it, alongside why the app is not asked to compute vectors on the phone. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
703 lines
24 KiB
Go
703 lines
24 KiB
Go
package repositories
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"log"
|
||
"nearle/db"
|
||
"nearle/models"
|
||
"nearle/utils"
|
||
"sort"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
/*
|
||
Scan-to-order reads from three places and this file is the only one that
|
||
knows which is which:
|
||
|
||
- the catalogue database (pgvector, one table per brand) — to turn a label
|
||
into a catalogue product;
|
||
- nearledb — who the customer is, which outlets they scanned into, and what
|
||
those outlets have on the shelf right now;
|
||
- Redis — a cache for the expensive and stable half (the label's vector and
|
||
its catalogue hits). Live stock is never cached.
|
||
|
||
Two connections are held rather than one because the catalogue must never be
|
||
reachable through the nearledb handle: the comment on db.CatalogueDB is
|
||
explicit about that and every catalogue reader in this package honours it.
|
||
*/
|
||
|
||
// CatalogueKey is how a tenant's product row points back at the catalogue.
|
||
// Imageid is the stable one; brand+catalogueid is kept for rows imported
|
||
// before imageid existed (see models.Products.Imageid).
|
||
type CatalogueKey struct {
|
||
Brand string
|
||
Catalogueid int64
|
||
Imageid string
|
||
}
|
||
|
||
// CatalogueHit is one catalogue row the search considered.
|
||
type CatalogueHit struct {
|
||
Brand string
|
||
ID int64
|
||
ProductName string
|
||
Title string
|
||
Category string
|
||
Size string
|
||
VariantKey string
|
||
ImageID string
|
||
ImageURL string
|
||
// Cosine distance from pgvector (0 = identical); -1 for a text-only hit.
|
||
Distance float64
|
||
}
|
||
|
||
// StoreOptionRow is one sellable product at one outlet, with its live stock.
|
||
type StoreOptionRow struct {
|
||
Tenantid int
|
||
Locationid int
|
||
Productid int
|
||
Productname string
|
||
Productbrand string
|
||
Catalogueid int64
|
||
Imageid string
|
||
Productimage string
|
||
Productunit string
|
||
Unitvalue string
|
||
Price float64
|
||
Stock int
|
||
// For a size row: the product it hangs under and the label given to it.
|
||
Parentid int
|
||
Variantname string
|
||
}
|
||
|
||
type ScanRepository interface {
|
||
// nearledb
|
||
CustomerExists(ctx context.Context, customerid int) (bool, error)
|
||
CustomerHome(ctx context.Context, customerid int) (lat, lng float64, ok bool, err error)
|
||
RegisteredStores(ctx context.Context, customerid int) ([]models.ScanStore, error)
|
||
// StoreOptions finds, at the given outlets, every published product tied
|
||
// to one of the catalogue keys (or, for hand-made products, one of the
|
||
// names) — and every size hanging under those products.
|
||
StoreOptions(ctx context.Context, locationids []int, keys []CatalogueKey, names []string) ([]StoreOptionRow, error)
|
||
// ProductAt is one product at one outlet with its live stock, or nil.
|
||
ProductAt(ctx context.Context, tenantid, locationid, productid int) (*StoreOptionRow, error)
|
||
|
||
// catalogue
|
||
VectorSearch(ctx context.Context, vector []float32, limit int) ([]CatalogueHit, error)
|
||
TextSearch(ctx context.Context, label string, limit int) ([]CatalogueHit, error)
|
||
VectorSearchAvailable() bool
|
||
// CatalogueRef is one product named by its catalogue key, with its other
|
||
// pack sizes after it. Nothing is recognised or scored.
|
||
CatalogueRef(ctx context.Context, brand string, id int64) ([]CatalogueHit, error)
|
||
|
||
// cache
|
||
CachedVector(ctx context.Context, model, label string) ([]float32, bool)
|
||
CacheVector(ctx context.Context, model, label string, v []float32)
|
||
CachedHits(ctx context.Context, method, label string) ([]CatalogueHit, bool)
|
||
CacheHits(ctx context.Context, method, label string, hits []CatalogueHit)
|
||
}
|
||
|
||
type scanRepository struct {
|
||
db *gorm.DB
|
||
catalogue *gorm.DB
|
||
|
||
// Catalogue tables and their columns, discovered once and refreshed on a
|
||
// timer — the catalogue pipeline adds brands without telling anyone.
|
||
tablesMu sync.Mutex
|
||
tables map[string]map[string]bool // table -> column set
|
||
tablesAt time.Time
|
||
embeddingDim int
|
||
|
||
// Process-local cache in front of Redis, bounded, so a hot label costs
|
||
// nothing even when Redis is not configured.
|
||
memMu sync.Mutex
|
||
memVecs map[string][]float32
|
||
memHits map[string][]CatalogueHit
|
||
}
|
||
|
||
const (
|
||
scanTablesTTL = 10 * time.Minute
|
||
scanVectorTTL = 7 * 24 * time.Hour // a label's vector never changes for a given model
|
||
scanHitsTTL = 30 * time.Minute // the catalogue is rebuilt by scrape; not for long
|
||
scanMemCacheMax = 2000
|
||
)
|
||
|
||
func NewScanRepository(nearle, catalogue *gorm.DB) ScanRepository {
|
||
return &scanRepository{
|
||
db: nearle,
|
||
catalogue: catalogue,
|
||
memVecs: make(map[string][]float32),
|
||
memHits: make(map[string][]CatalogueHit),
|
||
}
|
||
}
|
||
|
||
// ── nearledb ────────────────────────────────────────────────────────────────
|
||
|
||
func (r *scanRepository) CustomerExists(ctx context.Context, customerid int) (bool, error) {
|
||
var n int64
|
||
err := r.db.WithContext(ctx).Raw(
|
||
`SELECT COUNT(1) FROM customers WHERE customerid = ?`, customerid).Scan(&n).Error
|
||
return n > 0, err
|
||
}
|
||
|
||
// CustomerHome is the saved primary address, falling back to the customers
|
||
// row itself. Either may be blank or unparsable — a customer created from a
|
||
// phone number alone has neither — and that is reported as ok=false rather
|
||
// than as (0, 0), which is a real place in the Gulf of Guinea.
|
||
func (r *scanRepository) CustomerHome(ctx context.Context, customerid int) (float64, float64, bool, error) {
|
||
var row struct {
|
||
Lat string
|
||
Lng string
|
||
}
|
||
err := r.db.WithContext(ctx).Raw(`
|
||
SELECT COALESCE(NULLIF(l.latitude, ''), c.latitude, '') AS lat,
|
||
COALESCE(NULLIF(l.longitude, ''), c.longitude, '') AS lng
|
||
FROM customers c
|
||
LEFT JOIN customerlocations l ON l.customerid = c.customerid AND l.primaryaddress = 1
|
||
WHERE c.customerid = ?
|
||
LIMIT 1`, customerid).Scan(&row).Error
|
||
if err != nil {
|
||
return 0, 0, false, err
|
||
}
|
||
lat, lng, ok := utils.ParseLatLng(row.Lat, row.Lng)
|
||
return lat, lng, ok, nil
|
||
}
|
||
|
||
// RegisteredStores is every active outlet of every tenant the customer has
|
||
// scanned into. A tenantcustomers row with locationid 0 means "the tenant",
|
||
// i.e. all of its outlets; a non-zero one pins a single outlet.
|
||
func (r *scanRepository) RegisteredStores(ctx context.Context, customerid int) ([]models.ScanStore, error) {
|
||
var rows []struct {
|
||
Tenantid int
|
||
Tenantname string
|
||
Locationid int
|
||
Locationname string
|
||
Address string
|
||
Latitude string
|
||
Longitude string
|
||
Deliveryradius int
|
||
Deliverymins int
|
||
Opentime string
|
||
Closetime string
|
||
}
|
||
err := r.db.WithContext(ctx).Raw(`
|
||
SELECT DISTINCT
|
||
tl.tenantid, COALESCE(t.tenantname, '') AS tenantname,
|
||
tl.locationid, COALESCE(tl.locationname, '') AS locationname,
|
||
COALESCE(tl.address, '') AS address,
|
||
COALESCE(tl.latitude, '') AS latitude, COALESCE(tl.longitude, '') AS longitude,
|
||
COALESCE(tl.deliveryradius, 0) AS deliveryradius, COALESCE(tl.deliverymins, 0) AS deliverymins,
|
||
COALESCE(tl.opentime, '') AS opentime, COALESCE(tl.closetime, '') AS closetime
|
||
FROM tenantcustomers tc
|
||
INNER JOIN tenantlocations tl
|
||
ON tl.tenantid = tc.tenantid
|
||
AND (COALESCE(tc.locationid, 0) = 0 OR tc.locationid = tl.locationid)
|
||
LEFT JOIN tenants t ON t.tenantid = tl.tenantid
|
||
WHERE tc.customerid = ?
|
||
AND LOWER(COALESCE(tl.status, 'active')) <> 'inactive'
|
||
ORDER BY tl.tenantid, tl.locationid`, customerid).Scan(&rows).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
now := time.Now()
|
||
stores := make([]models.ScanStore, 0, len(rows))
|
||
for _, row := range rows {
|
||
lat, lng, _ := utils.ParseLatLng(row.Latitude, row.Longitude)
|
||
stores = append(stores, models.ScanStore{
|
||
Tenantid: row.Tenantid,
|
||
Tenantname: row.Tenantname,
|
||
Locationid: row.Locationid,
|
||
Locationname: row.Locationname,
|
||
Address: row.Address,
|
||
Latitude: lat,
|
||
Longitude: lng,
|
||
DistanceKm: -1,
|
||
Deliveryradius: row.Deliveryradius,
|
||
Deliverymins: row.Deliverymins,
|
||
Open: utils.OpenNow(row.Opentime, row.Closetime, now),
|
||
})
|
||
}
|
||
return stores, nil
|
||
}
|
||
|
||
// storeOptionSelect is the projection every outlet read shares, so the
|
||
// price and stock rules cannot differ between the lookup and the confirm.
|
||
//
|
||
// Price: the outlet's own price when it set one, else the tenant's retail
|
||
// price — the same rule GetProducts applies. Stock: the live IN−OUT balance
|
||
// of the ledger at that outlet, the same expression the app displays, so a
|
||
// product can never be offered here and show 0 on the next screen.
|
||
const storeOptionSelect = `
|
||
SELECT a.tenantid, b.locationid, a.productid,
|
||
COALESCE(a.productname, '') AS productname,
|
||
LOWER(COALESCE(a.productbrand, '')) AS productbrand,
|
||
COALESCE(a.catalogueid, 0) AS catalogueid,
|
||
COALESCE(a.imageid, '') AS imageid,
|
||
COALESCE(a.productimage, '') AS productimage,
|
||
COALESCE(a.productunit, '') AS productunit,
|
||
COALESCE(a.unitvalue, '') AS unitvalue,
|
||
CASE WHEN COALESCE(b.price, 0) > 0 THEN b.price ELSE COALESCE(a.retailprice, 0) END AS price,
|
||
COALESCE((
|
||
SELECT SUM(CASE WHEN LOWER(c.stocktype) = 'in' THEN c.quantity
|
||
WHEN LOWER(c.stocktype) = 'out' THEN -c.quantity
|
||
ELSE 0 END)
|
||
FROM productstocks c
|
||
WHERE c.productid = a.productid AND c.locationid = b.locationid AND c.tenantid = a.tenantid
|
||
), 0) AS stock,
|
||
COALESCE(v.productid, 0) AS parentid,
|
||
COALESCE(v.variantname, '') AS variantname
|
||
FROM products a
|
||
INNER JOIN productlocations b ON b.productid = a.productid AND b.tenantid = a.tenantid
|
||
LEFT JOIN productvariants v ON v.variantproductid = a.productid AND v.tenantid = a.tenantid
|
||
AND LOWER(COALESCE(v.status, 'active')) <> 'inactive'`
|
||
|
||
func (r *scanRepository) StoreOptions(ctx context.Context, locationids []int, keys []CatalogueKey, names []string) ([]StoreOptionRow, error) {
|
||
if len(locationids) == 0 || (len(keys) == 0 && len(names) == 0) {
|
||
return nil, nil
|
||
}
|
||
|
||
// The products that ARE the catalogue match, at these outlets.
|
||
var matchConds []string
|
||
var args []interface{}
|
||
args = append(args, locationids)
|
||
for _, k := range keys {
|
||
if k.Imageid != "" {
|
||
matchConds = append(matchConds, "a.imageid = ?")
|
||
args = append(args, k.Imageid)
|
||
}
|
||
if k.Brand != "" && k.Catalogueid > 0 {
|
||
matchConds = append(matchConds, "(LOWER(a.productbrand) = ? AND a.catalogueid = ?)")
|
||
args = append(args, strings.ToLower(k.Brand), k.Catalogueid)
|
||
}
|
||
}
|
||
for _, n := range names {
|
||
if n = strings.ToLower(strings.TrimSpace(n)); n != "" {
|
||
matchConds = append(matchConds, "LOWER(a.productname) = ?")
|
||
args = append(args, n)
|
||
}
|
||
}
|
||
if len(matchConds) == 0 {
|
||
return nil, nil
|
||
}
|
||
|
||
// Two reads rather than one recursive query: the second is keyed on the
|
||
// first's product ids, and a variant of a variant is not a thing here.
|
||
query := storeOptionSelect + `
|
||
WHERE a.approve = 1 AND b.publishedat IS NOT NULL
|
||
AND b.locationid IN (?)
|
||
AND (` + strings.Join(matchConds, " OR ") + `)`
|
||
|
||
var parents []StoreOptionRow
|
||
if err := r.db.WithContext(ctx).Raw(query, args...).Scan(&parents).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
if len(parents) == 0 {
|
||
return nil, nil
|
||
}
|
||
|
||
parentIDs := make([]int, 0, len(parents))
|
||
for _, p := range parents {
|
||
parentIDs = append(parentIDs, p.Productid)
|
||
}
|
||
|
||
// The sizes hanging under those products, at the same outlets. Only the
|
||
// rows whose parent is one of ours — the LEFT JOIN in the select can
|
||
// attach any parent, so it is pinned here.
|
||
var sizes []StoreOptionRow
|
||
err := r.db.WithContext(ctx).Raw(storeOptionSelect+`
|
||
WHERE a.approve = 1 AND b.publishedat IS NOT NULL
|
||
AND b.locationid IN (?)
|
||
AND v.productid IN (?)`, locationids, parentIDs).Scan(&sizes).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
return append(parents, sizes...), nil
|
||
}
|
||
|
||
func (r *scanRepository) ProductAt(ctx context.Context, tenantid, locationid, productid int) (*StoreOptionRow, error) {
|
||
var rows []StoreOptionRow
|
||
err := r.db.WithContext(ctx).Raw(storeOptionSelect+`
|
||
WHERE a.approve = 1 AND b.publishedat IS NOT NULL
|
||
AND a.tenantid = ? AND b.locationid = ? AND a.productid = ?
|
||
LIMIT 1`, tenantid, locationid, productid).Scan(&rows).Error
|
||
if err != nil || len(rows) == 0 {
|
||
return nil, err
|
||
}
|
||
return &rows[0], nil
|
||
}
|
||
|
||
// ── catalogue ───────────────────────────────────────────────────────────────
|
||
|
||
// brandTables is every `brand_*` table and its columns, cached briefly.
|
||
func (r *scanRepository) brandTables(ctx context.Context) (map[string]map[string]bool, error) {
|
||
if r.catalogue == nil {
|
||
return nil, ErrCatalogueDBUnavailable
|
||
}
|
||
r.tablesMu.Lock()
|
||
defer r.tablesMu.Unlock()
|
||
if r.tables != nil && time.Since(r.tablesAt) < scanTablesTTL {
|
||
return r.tables, nil
|
||
}
|
||
|
||
var rows []struct {
|
||
TableName string
|
||
ColumnName string
|
||
}
|
||
err := r.catalogue.WithContext(ctx).Raw(`
|
||
SELECT c.table_name, c.column_name
|
||
FROM information_schema.columns c
|
||
WHERE c.table_schema = 'public' AND c.table_name LIKE 'brand\_%'`).Scan(&rows).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
tables := make(map[string]map[string]bool)
|
||
for _, row := range rows {
|
||
if tables[row.TableName] == nil {
|
||
tables[row.TableName] = make(map[string]bool)
|
||
}
|
||
tables[row.TableName][row.ColumnName] = true
|
||
}
|
||
for name, cols := range tables {
|
||
if !cols["id"] || !cols["product_name"] {
|
||
delete(tables, name)
|
||
}
|
||
}
|
||
|
||
// The vector width, read from the first embedding column found. pgvector
|
||
// stores it as the type modifier, so a mismatch with the model can be
|
||
// named in the error instead of surfacing as a bare "different vector
|
||
// dimensions" from the driver.
|
||
if r.embeddingDim == 0 {
|
||
for name, cols := range tables {
|
||
if !cols["embedding"] {
|
||
continue
|
||
}
|
||
var dim int
|
||
r.catalogue.WithContext(ctx).Raw(`
|
||
SELECT a.atttypmod FROM pg_attribute a
|
||
JOIN pg_class c ON c.oid = a.attrelid
|
||
WHERE c.relname = ? AND a.attname = 'embedding'`, name).Scan(&dim)
|
||
if dim > 0 {
|
||
r.embeddingDim = dim
|
||
}
|
||
break
|
||
}
|
||
}
|
||
|
||
r.tables, r.tablesAt = tables, time.Now()
|
||
return tables, nil
|
||
}
|
||
|
||
// VectorSearchAvailable is whether any catalogue table carries a vector.
|
||
func (r *scanRepository) VectorSearchAvailable() bool {
|
||
tables, err := r.brandTables(context.Background())
|
||
if err != nil {
|
||
return false
|
||
}
|
||
for _, cols := range tables {
|
||
if cols["embedding"] {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
// hitColumns is the projection each search returns, with NULL stand-ins for
|
||
// columns a particular brand table lacks — the same tolerance
|
||
// catalogueRepository applies, for the same reason: a newer table missing
|
||
// one enrichment column is still a perfectly good catalogue of products.
|
||
func hitColumns(brand string, cols map[string]bool) string {
|
||
opt := func(name string) string {
|
||
if cols[name] {
|
||
return "COALESCE(" + name + ", '') AS " + name
|
||
}
|
||
return "'' AS " + name
|
||
}
|
||
return fmt.Sprintf(`'%s' AS brand, id, COALESCE(product_name, '') AS product_name, %s, %s, %s, %s, %s, %s`,
|
||
brand, opt("title"), opt("category"), opt("size"), opt("variant_key"), opt("image_id"), opt("image_url"))
|
||
}
|
||
|
||
// VectorSearch ranks every brand table by cosine distance to the label's
|
||
// vector and merges the top of each.
|
||
//
|
||
// One branch per table, each with its own ORDER BY and LIMIT inside
|
||
// parentheses, so Postgres can use the per-table vector index instead of
|
||
// scanning the union. The literal is bound as a parameter and cast — never
|
||
// concatenated — and table names come from information_schema, never from
|
||
// the request.
|
||
func (r *scanRepository) VectorSearch(ctx context.Context, vector []float32, limit int) ([]CatalogueHit, error) {
|
||
tables, err := r.brandTables(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if r.embeddingDim > 0 && len(vector) != r.embeddingDim {
|
||
return nil, fmt.Errorf("embedding is %d wide but the catalogue's embedding column is %d: EMBEDDING_MODEL/EMBEDDING_DIMENSIONS do not match the model that indexed the catalogue", len(vector), r.embeddingDim)
|
||
}
|
||
|
||
literal := utils.VectorLiteral(vector)
|
||
var branches []string
|
||
var args []interface{}
|
||
for _, table := range sortedKeys(tables) {
|
||
cols := tables[table]
|
||
if !cols["embedding"] {
|
||
continue
|
||
}
|
||
brand := strings.TrimPrefix(table, "brand_")
|
||
branches = append(branches, fmt.Sprintf(
|
||
`(SELECT %s, (embedding <=> ?::vector) AS distance FROM %s WHERE embedding IS NOT NULL ORDER BY embedding <=> ?::vector LIMIT %d)`,
|
||
hitColumns(brand, cols), table, limit))
|
||
args = append(args, literal, literal)
|
||
}
|
||
if len(branches) == 0 {
|
||
return nil, errors.New("no catalogue table has an embedding column")
|
||
}
|
||
|
||
query := strings.Join(branches, " UNION ALL ") + fmt.Sprintf(" ORDER BY distance LIMIT %d", limit)
|
||
var hits []CatalogueHit
|
||
if err := r.catalogue.WithContext(ctx).Raw(query, args...).Scan(&hits).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
return hits, nil
|
||
}
|
||
|
||
// TextSearch is the fallback when there is no embedder, and the tie-breaker
|
||
// beside it when there is: rows whose name or title contains the label, or
|
||
// contains every word of it.
|
||
func (r *scanRepository) TextSearch(ctx context.Context, label string, limit int) ([]CatalogueHit, error) {
|
||
tables, err := r.brandTables(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
label = strings.ToLower(strings.TrimSpace(label))
|
||
tokens := utils.SearchTokens(label)
|
||
if label == "" || len(tokens) == 0 {
|
||
return nil, nil
|
||
}
|
||
|
||
var branches []string
|
||
var args []interface{}
|
||
for _, table := range sortedKeys(tables) {
|
||
cols := tables[table]
|
||
brand := strings.TrimPrefix(table, "brand_")
|
||
|
||
hay := "LOWER(COALESCE(product_name, ''))"
|
||
if cols["title"] {
|
||
hay = "LOWER(COALESCE(product_name, '') || ' ' || COALESCE(title, ''))"
|
||
}
|
||
if cols["search_query"] {
|
||
hay = "LOWER(COALESCE(product_name, '') || ' ' || COALESCE(title, '') || ' ' || COALESCE(search_query, ''))"
|
||
}
|
||
|
||
conds := []string{hay + " LIKE ?"}
|
||
args = append(args, "%"+label+"%")
|
||
all := make([]string, 0, len(tokens))
|
||
for _, tok := range tokens {
|
||
all = append(all, hay+" LIKE ?")
|
||
args = append(args, "%"+tok+"%")
|
||
}
|
||
conds = append(conds, "("+strings.Join(all, " AND ")+")")
|
||
|
||
branches = append(branches, fmt.Sprintf(
|
||
`(SELECT %s, -1::float8 AS distance FROM %s WHERE %s LIMIT %d)`,
|
||
hitColumns(brand, cols), table, strings.Join(conds, " OR "), limit))
|
||
}
|
||
if len(branches) == 0 {
|
||
return nil, nil
|
||
}
|
||
|
||
var hits []CatalogueHit
|
||
if err := r.catalogue.WithContext(ctx).Raw(strings.Join(branches, " UNION ALL "), args...).Scan(&hits).Error; err != nil {
|
||
return nil, err
|
||
}
|
||
return hits, nil
|
||
}
|
||
|
||
// tableFor resolves a brand the caller named to a real catalogue table.
|
||
//
|
||
// The lookup is against the tables discovered from information_schema, never
|
||
// a string built from the request: table names cannot be parameterised in
|
||
// SQL, so the discovered map is what keeps this from being an injection
|
||
// point. Both the table suffix ("britannia") and a display name ("24 Mantra"
|
||
// → brand_24_mantra) resolve.
|
||
func (r *scanRepository) tableFor(ctx context.Context, brand string) (string, map[string]bool, error) {
|
||
tables, err := r.brandTables(ctx)
|
||
if err != nil {
|
||
return "", nil, err
|
||
}
|
||
for _, candidate := range []string{
|
||
"brand_" + strings.ToLower(strings.TrimSpace(brand)),
|
||
"brand_" + normaliseBrandKey(brand),
|
||
} {
|
||
if cols, ok := tables[candidate]; ok {
|
||
return candidate, cols, nil
|
||
}
|
||
}
|
||
return "", nil, ErrUnknownBrand
|
||
}
|
||
|
||
// CatalogueRef reads one product by (brand, id) and appends its other pack
|
||
// sizes — same variant_key where the catalogue assigned one, same name
|
||
// otherwise, matching how the search groups a family.
|
||
//
|
||
// Distance is 0 on every row: nothing here was ranked, the caller said which
|
||
// product they meant.
|
||
func (r *scanRepository) CatalogueRef(ctx context.Context, brand string, id int64) ([]CatalogueHit, error) {
|
||
table, cols, err := r.tableFor(ctx, brand)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
suffix := strings.TrimPrefix(table, "brand_")
|
||
columns := hitColumns(suffix, cols)
|
||
|
||
var self []CatalogueHit
|
||
err = r.catalogue.WithContext(ctx).Raw(fmt.Sprintf(
|
||
`SELECT %s, 0::float8 AS distance FROM %s WHERE id = ?`, columns, table), id).Scan(&self).Error
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(self) == 0 {
|
||
return nil, nil
|
||
}
|
||
|
||
var siblings []CatalogueHit
|
||
if cols["variant_key"] && strings.TrimSpace(self[0].VariantKey) != "" {
|
||
err = r.catalogue.WithContext(ctx).Raw(fmt.Sprintf(
|
||
`SELECT %s, 0::float8 AS distance FROM %s WHERE variant_key = ? AND id <> ? ORDER BY id`,
|
||
columns, table), self[0].VariantKey, id).Scan(&siblings).Error
|
||
} else {
|
||
err = r.catalogue.WithContext(ctx).Raw(fmt.Sprintf(
|
||
`SELECT %s, 0::float8 AS distance FROM %s WHERE LOWER(product_name) = LOWER(?) AND id <> ? ORDER BY id`,
|
||
columns, table), self[0].ProductName, id).Scan(&siblings).Error
|
||
}
|
||
if err != nil {
|
||
// The product itself was found; losing its other sizes is the smaller
|
||
// failure and the caller asked for this one.
|
||
log.Printf("scan: could not read pack sizes of %s#%d: %v", brand, id, err)
|
||
return self, nil
|
||
}
|
||
return append(self, siblings...), nil
|
||
}
|
||
|
||
func sortedKeys(m map[string]map[string]bool) []string {
|
||
keys := make([]string, 0, len(m))
|
||
for k := range m {
|
||
keys = append(keys, k)
|
||
}
|
||
sort.Strings(keys)
|
||
return keys
|
||
}
|
||
|
||
// ── cache ───────────────────────────────────────────────────────────────────
|
||
//
|
||
// Two tiers. Redis is shared across replicas and survives a restart; the
|
||
// in-process map is there so the request after a cache hit costs no network
|
||
// round trip at all, and so a deployment without Redis still gets the
|
||
// benefit within one process. Neither tier ever holds stock.
|
||
|
||
func scanCacheKey(kind, scope, label string) string {
|
||
sum := sha256.Sum256([]byte(strings.ToLower(strings.TrimSpace(label))))
|
||
return "scan:" + kind + ":v1:" + scope + ":" + hex.EncodeToString(sum[:16])
|
||
}
|
||
|
||
func (r *scanRepository) CachedVector(ctx context.Context, model, label string) ([]float32, bool) {
|
||
key := scanCacheKey("emb", model, label)
|
||
|
||
r.memMu.Lock()
|
||
v, ok := r.memVecs[key]
|
||
r.memMu.Unlock()
|
||
if ok {
|
||
return v, true
|
||
}
|
||
|
||
if db.Rdb == nil {
|
||
return nil, false
|
||
}
|
||
raw, err := db.Rdb.Get(ctx, key).Bytes()
|
||
if err != nil {
|
||
return nil, false
|
||
}
|
||
if json.Unmarshal(raw, &v) != nil || len(v) == 0 {
|
||
return nil, false
|
||
}
|
||
r.remember(key, v, nil)
|
||
return v, true
|
||
}
|
||
|
||
func (r *scanRepository) CacheVector(ctx context.Context, model, label string, v []float32) {
|
||
key := scanCacheKey("emb", model, label)
|
||
r.remember(key, v, nil)
|
||
if db.Rdb == nil {
|
||
return
|
||
}
|
||
if raw, err := json.Marshal(v); err == nil {
|
||
if err := db.Rdb.Set(ctx, key, raw, scanVectorTTL).Err(); err != nil {
|
||
log.Printf("scan: could not cache vector: %v", err)
|
||
}
|
||
}
|
||
}
|
||
|
||
func (r *scanRepository) CachedHits(ctx context.Context, method, label string) ([]CatalogueHit, bool) {
|
||
key := scanCacheKey("hits", method, label)
|
||
|
||
r.memMu.Lock()
|
||
h, ok := r.memHits[key]
|
||
r.memMu.Unlock()
|
||
if ok {
|
||
return h, true
|
||
}
|
||
|
||
if db.Rdb == nil {
|
||
return nil, false
|
||
}
|
||
raw, err := db.Rdb.Get(ctx, key).Bytes()
|
||
if err != nil {
|
||
return nil, false
|
||
}
|
||
if json.Unmarshal(raw, &h) != nil {
|
||
return nil, false
|
||
}
|
||
r.remember(key, nil, h)
|
||
return h, true
|
||
}
|
||
|
||
func (r *scanRepository) CacheHits(ctx context.Context, method, label string, hits []CatalogueHit) {
|
||
key := scanCacheKey("hits", method, label)
|
||
r.remember(key, nil, hits)
|
||
if db.Rdb == nil {
|
||
return
|
||
}
|
||
if raw, err := json.Marshal(hits); err == nil {
|
||
if err := db.Rdb.Set(ctx, key, raw, scanHitsTTL).Err(); err != nil {
|
||
log.Printf("scan: could not cache hits: %v", err)
|
||
}
|
||
}
|
||
}
|
||
|
||
// remember writes one entry into the process-local tier. Eviction is the
|
||
// simplest thing that bounds memory: when full, drop everything. Labels are
|
||
// short-lived popularity, not a working set worth an LRU.
|
||
func (r *scanRepository) remember(key string, v []float32, h []CatalogueHit) {
|
||
r.memMu.Lock()
|
||
defer r.memMu.Unlock()
|
||
if len(r.memVecs)+len(r.memHits) >= scanMemCacheMax {
|
||
r.memVecs = make(map[string][]float32)
|
||
r.memHits = make(map[string][]CatalogueHit)
|
||
}
|
||
if v != nil {
|
||
r.memVecs[key] = v
|
||
}
|
||
if h != nil {
|
||
r.memHits[key] = h
|
||
}
|
||
}
|