main had moved on with retrieval work validated against real queries — minTokenHits (the word match needs two thirds of the label, not all of it), separator folding so "Parle G"/"Parle-G"/"ParleG" all reach Parle-G, the floor at 0.50 after "Paracetamol" came back as "Paneer Makhni 500ml" at 0.304, and ties broken on cosine distance instead of name. All of that is kept exactly as it was. The conflict was in textScore: this branch replaced the substring rule with a coverage formula to stop a bare brand name resolving to one arbitrary product. That is the wrong half to change. The substring rule scores every product of a brand 0.95 IDENTICALLY, and that tie is not the bug — it is the signal. isAmbiguous reads it, so the branch's coverage rewrite is dropped and the ambiguity layer alone does the work: "britannia" → all 258 rows tie at 0.95 → ambiguous: true + candidates "Parle G" → folding and the single-character token still land it a real name → runner-up far behind → match, unchanged Dropped with it: scanSpecificEnough, the per-hit text score, and the proportional confirmation bonus — the flat +0.10 is back. Simpler, and it leaves main's tuning untouched. TestTextScoreRewardsSpecificityNotJustOverlap tested the removed formula and is replaced by TestABrandNameScoresItsProductsIdentically, which guards the tie itself: a formula that broke it on name length or word count would bring the bug back. Docs carry both rationales, and now say plainly that confidence stays high on the ambiguous path — gate on `ambiguous`, never on `confidence`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
730 lines
25 KiB
Go
730 lines
25 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
|
||
}
|
||
|
||
// minTokenHits is how many of the label's words a row must carry to be worth
|
||
// looking at. Every word was once required, which meant a single word the
|
||
// catalogue does not use — "Parle G biscuit pack", "Milk Bikis pack" — kept
|
||
// the right product out of the result entirely, leaving the vector search to
|
||
// answer alone and confidently wrong. Most of them is enough; scoring sorts
|
||
// out the rest.
|
||
func minTokenHits(n int) int {
|
||
if n <= 2 {
|
||
return n
|
||
}
|
||
return (n*2 + 2) / 3 // two thirds, rounded up; never below 2 for n >= 3
|
||
}
|
||
|
||
// 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
|
||
// carry most of its words.
|
||
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, ''))"
|
||
}
|
||
|
||
// How well a row matches, as a number: the whole label as a substring
|
||
// outweighs any number of loose words, then one point per word found.
|
||
hits := make([]string, 0, len(tokens)+1)
|
||
hits = append(hits, "(CASE WHEN "+hay+" LIKE ? THEN 100 ELSE 0 END)")
|
||
for range tokens {
|
||
hits = append(hits, "(CASE WHEN "+hay+" LIKE ? THEN 1 ELSE 0 END)")
|
||
}
|
||
rank := strings.Join(hits, " + ")
|
||
|
||
// The expression appears twice in the SQL — once to filter, once to
|
||
// order — so its arguments are bound twice, in that order.
|
||
bind := func() {
|
||
args = append(args, "%"+label+"%")
|
||
for _, tok := range tokens {
|
||
args = append(args, "%"+tok+"%")
|
||
}
|
||
}
|
||
bind()
|
||
bind()
|
||
|
||
// Ordering matters as much as the threshold: a looser WHERE lets more
|
||
// rows qualify, and an unordered LIMIT would then be free to return
|
||
// the wrong ones. Best match per brand first, id to keep it stable.
|
||
branches = append(branches, fmt.Sprintf(
|
||
`(SELECT %s, -1::float8 AS distance FROM %s WHERE (%s) >= %d ORDER BY (%s) DESC, id LIMIT %d)`,
|
||
hitColumns(brand, cols), table, rank, minTokenHits(len(tokens)), rank, 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
|
||
}
|
||
}
|