backend_fiesta/repositories/scanRepository.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

	// 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
}

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
	}
}