Behavision/server/internal/contract/contract.go

// Package contract is the wire format between a store agent and the server.
//
// Written down rather than left to emerge from whichever struct happened to be
// convenient: the agent ships to customer sites and cannot be redeployed on
// demand, so the server must keep parsing what old agents send. Every field
// added here must be optional, and no field may change meaning.
//
// Topics
//
//	bv/<client>.<site>/visit      agent -> server, one recognised person
//	bv/<client>.<site>/heartbeat  agent -> server, "I am alive"
//	bv/<client>.<site>/status     agent -> server, health and pipeline stats
//	bv/<client>.<site>/cmd/#      server -> agent
//
// The prefix is the agent's MQTT username, which the broker enforces with
// `pattern write bv/%u/...`. A site therefore cannot publish under another
// site's prefix even if it tries, and the server does not have to trust the
// topic - it re-derives the tenant from the username it was published under.
package contract

import (
	"errors"
	"fmt"
	"strings"
	"time"
)

// EmbeddingDim is fixed by the encoder. A payload with any other length is
// rejected rather than stored: a wrong-length vector cannot be compared with
// anything and would sit in the gallery poisoning every future search.
const EmbeddingDim = 512

// Visit is one person seen at one site.
type Visit struct {
	// EventID is generated by the agent and is the idempotency key. MQTT
	// delivery is at-least-once by design, so a reconnect can redeliver; without
	// this a store's footfall silently doubles, and footfall is the number the
	// customer is paying for.
	EventID string `json:"event_id"`

	OccurredAt time.Time `json:"occurred_at"`
	CameraID   string    `json:"camera_id"`

	// IsNew is the agent's local verdict: did its own gallery already know this
	// face. Advisory only — the server re-decides against the client-wide
	// gallery, because a person new to this store may be known at another one.
	IsNew      bool    `json:"is_new"`
	Similarity float32 `json:"similarity"`
	Quality    float32 `json:"quality"`

	// LocalVisitorID ties this event back to the identity in the agent's own
	// SQLite, so a support question ("which local record is this?") is
	// answerable without guessing.
	LocalVisitorID int64 `json:"local_visitor_id,omitempty"`

	// Embedding is biometric personal data. Optional: a site may be configured
	// to keep templates local and send only counts.
	Embedding []float32 `json:"embedding,omitempty"`
	Model     string    `json:"model,omitempty"`

	// ImageKey is an object-store KEY, never a URL. A stored URL is permanent
	// and unrevocable; a key is presigned on read and expires.
	ImageKey string `json:"image_key,omitempty"`

	Attributes map[string]any `json:"attributes,omitempty"`
}

// Validate rejects what cannot be stored meaningfully.
//
// Returns two kinds of failure deliberately: a permanent one means the message
// will never become valid, so the consumer must drop it rather than retry
// forever — one bad message at the head of a queue must not stop every good one
// behind it.
func (v *Visit) Validate() error {
	if strings.TrimSpace(v.EventID) == "" {
		return fmt.Errorf("%w: event_id is required (idempotency key)", ErrPermanent)
	}
	if len(v.EventID) > 128 {
		return fmt.Errorf("%w: event_id too long", ErrPermanent)
	}
	if v.OccurredAt.IsZero() {
		return fmt.Errorf("%w: occurred_at is required", ErrPermanent)
	}
	// A clock skewed years into the future would park a visit at the top of
	// every "recent" report forever. Reject rather than clamp: silently moving
	// a timestamp makes the footfall report a lie that looks fine.
	if v.OccurredAt.After(time.Now().Add(24 * time.Hour)) {
		return fmt.Errorf("%w: occurred_at is more than a day in the future "+
			"(%s) - check the site's clock", ErrPermanent, v.OccurredAt)
	}
	if len(v.Embedding) > 0 {
		if len(v.Embedding) != EmbeddingDim {
			return fmt.Errorf("%w: embedding has %d dimensions, want %d",
				ErrPermanent, len(v.Embedding), EmbeddingDim)
		}
		if v.Model == "" {
			// Vectors from different encoders occupy different spaces and must
			// never be compared. An untagged one cannot be safely stored.
			return fmt.Errorf("%w: embedding sent without a model tag", ErrPermanent)
		}
	}
	return nil
}

// Heartbeat says a site is alive.
//
// Without it "this site is switched off" and "this shop had no customers" are
// the same row of zeroes on the customer's report, and only one of them is
// something to act on.
type Heartbeat struct {
	SentAt           time.Time       `json:"sent_at"`
	AgentVersion     string          `json:"agent_version,omitempty"`
	EngineVersion    string          `json:"engine_version,omitempty"`
	RecognitionModel string          `json:"recognition_model,omitempty"`
	Cameras          map[string]bool `json:"cameras,omitempty"`
	// Queued and Dropped come from the agent's spool. Dropped is non-zero only
	// when a site was offline long enough to overflow its queue, which means
	// that site genuinely lost footfall - it must be visible, not inferred.
	Queued  int    `json:"queued,omitempty"`
	Dropped uint64 `json:"dropped,omitempty"`

	// FractionBelowGate is the share of faces this site's cameras saw that fell
	// under the enrolment gate — the number that decides whether a footfall
	// figure can be believed at all. It travels on the heartbeat rather than on
	// each visit because it describes the SITE, and because the visits it is
	// about are precisely the ones that never became visits.
	//
	// Optional: an old agent sends nothing and the server keeps the last value
	// it had rather than recording a perfect zero it was never told.
	FractionBelowGate float32 `json:"fraction_below_gate,omitempty"`
}

var (
	// ErrPermanent means the message will never be valid. Drop it.
	ErrPermanent = errors.New("permanent")
	// ErrTransient means try again later.
	ErrTransient = errors.New("transient")
)

// Topic is a parsed agent topic.
type Topic struct {
	Username string // "<client>.<site>", as the broker authenticated it
	Client   string
	Site     string
	Kind     string // visit | heartbeat | status | cmd
	Rest     string
}

// ParseTopic splits bv/<client>.<site>/<kind>[/...].
func ParseTopic(topic string) (Topic, error) {
	parts := strings.Split(strings.Trim(topic, "/"), "/")
	if len(parts) < 3 || parts[0] != "bv" {
		return Topic{}, fmt.Errorf("%w: unexpected topic %q", ErrPermanent, topic)
	}
	user := parts[1]
	// Exactly one dot: "acme.store1". Splitting on the first dot would let
	// "acme.store.1" through as client "acme", site "store.1", which is a
	// different site than the one the broker authenticated.
	dot := strings.Split(user, ".")
	if len(dot) != 2 || dot[0] == "" || dot[1] == "" {
		return Topic{}, fmt.Errorf("%w: topic %q has no <client>.<site> segment",
			ErrPermanent, topic)
	}
	return Topic{
		Username: user,
		Client:   dot[0],
		Site:     dot[1],
		Kind:     parts[2],
		Rest:     strings.Join(parts[3:], "/"),
	}, nil
}