738 lines
26 KiB
Go
738 lines
26 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
|
||
// The shop's own cost, distinct from Price. Selected because the product
|
||
// screens already return it and the app asked for it by name.
|
||
Productcost float64
|
||
Categoryid int
|
||
Subcategoryid int
|
||
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(a.productcost, 0) AS productcost,
|
||
COALESCE(a.categoryid, 0) AS categoryid,
|
||
COALESCE(a.subcategoryid, 0) AS subcategoryid,
|
||
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
|
||
}
|
||
}
|