Loyaly/Behavision
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 22 Wiki Activity

Live view at head office, relayed through the agent's outbound connection

Browse Source 
I got this wrong first time. "Head office cannot show live video cheaply"
conflated TRUE VIDEO with SEEING THE CAMERA NOW, and only the first needs
WebRTC and a TURN server.

The shop PC is behind a router with no inbound route, so head office
cannot pull the engine's MJPEG. It can answer the agent's outbound
requests, which is the shape of everything else here: the server holds a
poll open, the agent asks "is anyone watching?", and pushes JPEGs up for
exactly as long as somebody is.

Measured on the office camera: 98 KB full frame, 20.8 KB re-encoded at
640/q60, so one watcher costs ~83 KB/s. 47 frames arrived in 12 seconds -
4 fps, as configured. The UI says "about 4 frames a second" rather than
letting anyone conclude the camera stutters.

Nothing is uploaded when nobody is looking, which is the whole cost
argument: Publish returns false once the last viewer goes, interest lapses
on a timer each viewer refreshes as it reads (so a closed tab stops the
upload within seconds), one push is capped at five minutes, and the UI
streams one camera at a time.

LiveHub is deliberately the opposite of the arrivals Hub. There a doorbell
pushes nothing because nothing may be lost; here a dropped frame is the
correct outcome, so each viewer has a one-slot buffer that is overwritten -
the only frame worth having is the newest, and a queue would show an
ever-growing delay behind the shop instead of dropping back to live.

Ownership is proved once, before anything streams: the relay is keyed on a
camera id, a hub does not know whose camera it holds, and a camera id is
not a secret. Verified: another tenant gets 404, no session gets 401, and
an agent cannot push into another site's camera.

Also fixes a bug I introduced with it - the Live button was gated on
`connected`, which is head office's last report and up to two minutes
stale, so it hid itself during every reconnect. "Is that camera really
down?" is exactly when somebody wants to look, and a hidden control says
"you cannot" where the honest answer is "here is why".

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HViLj9gYNRtSr7YVZmW5sn
This commit is contained in:
Suriyakumarvijayanayagam
2026-09-04 16:46:23 +05:30
parent 2cd7a78ddc
commit 18686cbceb
23 changed files with 1250 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
server/internal/api/api.go
View File

@@ -65,6 +65,14 @@ type Store interface {
// reachable only with that site's own agent token.
AgentCameras(ctx context.Context, siteID string) ([]AgentCamera, error)
ApplyAgentReport(ctx context.Context, clientID, siteID string, rep AgentCameraReport) error
// CameraRef resolves one of a TENANT's cameras to its site and the name the
// engine knows it by. Used to prove ownership before anything is streamed.
CameraRef(ctx context.Context, clientID, cameraID string) (siteID, engineID string, err error)
// CameraRefBySite is the same question asked by an agent, which is
// authenticated for a site rather than a tenant.
CameraRefBySite(ctx context.Context, siteID, cameraID string) (site, engineID string, err error)
// SiteCameraIDs lists a site's camera uuids, for the agent's live poll.
SiteCameraIDs(ctx context.Context, siteID string) ([]string, error)
// Camera pictures held by this server, for deployments with no object
// storage. Where a bucket is configured neither of these is called.
PutCameraSnapshot(ctx context.Context, clientID, siteID, cameraID string, jpeg []byte) error
@@ -118,6 +126,10 @@ type Server struct {
// business questions the screens ask. Nil means this deployment has no
// API key, which is supported: the UI hides the panel.
Assistant Assistant
// Live relays camera frames from a shop PC to whoever is watching, on
// demand. Created on first use.
Live *LiveHub
liveOnce sync.Once
// Hub wakes live arrival streams when the MQTT consumer records a visit.
// Nil is supported and means the streams fall back to their slow tick -
// a server assembled without one is slower, not broken.
@@ -197,6 +209,7 @@ func (s *Server) Routes() *http.ServeMux {
mux.HandleFunc("PATCH /api/cameras/{id}", s.authed(s.handleUpdateCamera))
mux.HandleFunc("DELETE /api/cameras/{id}", s.authed(s.handleDeleteCamera))
mux.HandleFunc("GET /api/cameras/{id}/snapshot.jpg", s.authed(s.handleGetSnapshot))
mux.HandleFunc("GET /api/cameras/{id}/live", s.authed(s.handleWatchLive))
// Prove a camera works: "connection" asks whether the shop PC can open the
// stream, "placement" asks whether somebody walking past produces a view
// good enough to recognise. Two questions, because a camera passes the
@@ -242,6 +255,9 @@ func (s *Server) Routes() *http.ServeMux {
mux.HandleFunc("POST /api/agent/cameras", s.agentAuthed(s.handleAgentCameraReport))
mux.HandleFunc("PUT /api/agent/cameras/{camera}/snapshot",
s.agentAuthed(s.handlePutSnapshot))
mux.HandleFunc("GET /api/agent/live", s.agentAuthed(s.handleAgentLiveWanted))
mux.HandleFunc("POST /api/agent/cameras/{camera}/live",
s.agentAuthed(s.handleAgentPushLive))
mux.HandleFunc("GET /api/agent/checks", s.agentAuthed(s.handleAgentChecks))
mux.HandleFunc("POST /api/agent/checks", s.agentAuthed(s.handleAgentCheckResult))

51
server/internal/api/fake_test.go
View File

@@ -19,6 +19,11 @@ import (
// live - which tenant, which message on failure, what is echoed back - and
// those are exactly what a real database would make slow and awkward to test.
type fakeStore struct {
// Which tenant and site each camera belongs to. The live relay is keyed on
// a camera id and a hub does not know whose camera it holds, so ownership
// is proved before anything streams - and that is what these tests check.
cameraRefs map[string]cameraRef
// Camera pictures held by the server, for a deployment with no bucket.
// Keyed as written by PutCameraSnapshot (by camera_id) and as read by
// CameraSnapshot ("client/camera"), so a test has to say which it means.
@@ -622,3 +627,49 @@ func (f *fakeStore) CameraSnapshot(_ context.Context, clientID, cameraID string)
}
return img, time.Unix(1756900000, 0).UTC(), nil
}
// ------------------------------------------------------------ live relay --
func (f *fakeStore) CameraRef(_ context.Context, clientID, cameraID string) (string, string, error) {
f.mu.Lock()
defer f.mu.Unlock()
ref, ok := f.cameraRefs[cameraID]
if !ok || ref.client != clientID {
return "", "", ErrNoSnapshot
}
return ref.site, ref.engineID, nil
}
func (f *fakeStore) CameraRefBySite(_ context.Context, siteID, cameraID string) (string, string, error) {
f.mu.Lock()
defer f.mu.Unlock()
ref, ok := f.cameraRefs[cameraID]
if !ok || ref.site != siteID {
return "", "", ErrNoSnapshot
}
return ref.site, ref.engineID, nil
}
func (f *fakeStore) SiteCameraIDs(_ context.Context, siteID string) ([]string, error) {
f.mu.Lock()
defer f.mu.Unlock()
var out []string
for id, ref := range f.cameraRefs {
if ref.site == siteID {
out = append(out, id)
}
}
return out, nil
}
// addCameraRef registers a camera so ownership checks have something to check.
func (f *fakeStore) addCameraRef(id, client, site, engineID string) {
f.mu.Lock()
defer f.mu.Unlock()
if f.cameraRefs == nil {
f.cameraRefs = map[string]cameraRef{}
}
f.cameraRefs[id] = cameraRef{client: client, site: site, engineID: engineID}
}
type cameraRef struct{ client, site, engineID string }

173
server/internal/api/handlers_live.go Normal file
View File

@@ -0,0 +1,173 @@
package api
import (
"encoding/base64"
"encoding/binary"
"errors"
"fmt"
"io"
"net/http"
"time"
)
const (
// liveMaxFrame bounds one frame. The agent re-encodes to ~640 px before
// sending, so a frame is tens of KB; 1 MB is a ceiling, not a target.
liveMaxFrame = 1 << 20
// liveSession caps one push. A browser tab left open for a week must not
// leave a shop uploading for a week; the agent simply asks again while
// anyone is still watching, so the cap costs a reconnect, not the stream.
liveSession = 5 * time.Minute
// liveWaitForWork is how long the agent's poll is held open. Long enough
// that an idle site makes ~2 requests a minute; short enough to sit well
// inside any proxy's idle timeout.
liveWaitForWork = 25 * time.Second
)
// handleWatchLive streams one camera's frames to a signed-in user over SSE.
//
// SSE rather than serving MJPEG directly, for the same reason the snapshot is
// not a signed link: an <img> cannot send an Authorization header, and minting
// a URL that works without a session - for LIVE video of a shop floor, no less
// - would be a much worse trade than the 33% base64 costs.
func (s *Server) handleWatchLive(w http.ResponseWriter, r *http.Request) {
p := PrincipalFrom(r.Context())
id := r.PathValue("id")
if !looksLikeUUID(id) {
writeErr(w, http.StatusNotFound, "not_found", "No such camera.")
return
}
// Ownership is checked HERE, once, before anything is streamed. Everything
// after this point is keyed on a camera id, and a hub does not know whose
// camera it is holding.
if _, _, err := s.Store.CameraRef(r.Context(), p.ClientID, id); err != nil {
writeErr(w, http.StatusNotFound, "not_found", "No such camera.")
return
}
flusher, ok := w.(http.Flusher)
if !ok {
s.serverError(w, "live", errors.New("this server cannot stream"))
return
}
frames, release := s.live().Watch(id)
defer release()
h := w.Header()
h.Set("Content-Type", "text/event-stream")
h.Set("Cache-Control", "no-store")
h.Set("Connection", "keep-alive")
// Without this a proxy buffers the stream into one response that arrives
// when the connection closes - which for live video means never.
h.Set("X-Accel-Buffering", "no")
w.WriteHeader(http.StatusOK)
// Told up front, so a viewer can say "waiting for the shop PC" rather than
// showing an empty box while the agent is still being asked.
fmt.Fprint(w, "event: waiting\ndata: {}\n\n")
flusher.Flush()
ctx := r.Context()
// Refreshed as we go rather than once at the start: this is what tells the
// agent somebody is still there, and a viewer that has gone away stops a
// shop uploading within seconds without having to announce anything.
keep := time.NewTicker(liveIdle / 3)
defer keep.Stop()
for {
select {
case <-ctx.Done():
return
case <-keep.C:
s.live().Keep(id)
case frame, ok := <-frames:
if !ok {
return
}
s.live().Keep(id)
if _, err := fmt.Fprintf(w, "event: frame\ndata: %s\n\n",
base64.StdEncoding.EncodeToString(frame)); err != nil {
return
}
flusher.Flush()
}
}
}
// handleAgentLiveWanted is the shop PC asking whether anyone is watching.
//
// Held open rather than answered immediately: an agent polling every few
// seconds would put a floor under how quickly a live view can start, and one
// polling slowly would put a ceiling on it. Holding the request means pressing
// "Live" reaches the shop PC at once, and an idle site costs about two requests
// a minute.
func (s *Server) handleAgentLiveWanted(w http.ResponseWriter, r *http.Request, ap AgentPrincipal) {
ids, err := s.Store.SiteCameraIDs(r.Context(), ap.SiteID)
if err != nil {
s.serverError(w, "live wanted", err)
return
}
if wanted := s.live().WantedAmong(ids); len(wanted) > 0 {
writeJSON(w, http.StatusOK, map[string]any{"cameras": wanted})
return
}
select {
case <-r.Context().Done():
return
case <-s.live().Bell(ids):
case <-time.After(liveWaitForWork):
}
writeJSON(w, http.StatusOK, map[string]any{
"cameras": s.live().WantedAmong(ids)})
}
// handleAgentPushLive receives frames for as long as somebody is watching.
//
// One request carrying many frames, each prefixed with its length, rather than
// a request per frame: at a few frames a second the per-request overhead and
// the TLS handshakes would cost more than the pictures.
func (s *Server) handleAgentPushLive(w http.ResponseWriter, r *http.Request, ap AgentPrincipal) {
id := r.PathValue("camera")
if !looksLikeUUID(id) {
writeErr(w, http.StatusNotFound, "not_found", "No such camera.")
return
}
// The camera must belong to the AGENT's own site. Without this an agent
// could push its own pictures into another site's live view - a camera id
// is not a secret, and the agent supplies this one.
siteID, _, err := s.Store.CameraRefBySite(r.Context(), ap.SiteID, id)
if err != nil || siteID != ap.SiteID {
writeErr(w, http.StatusNotFound, "not_found", "No such camera.")
return
}
deadline := time.Now().Add(liveSession)
body := r.Body
var header [4]byte
frames := 0
for {
if time.Now().After(deadline) {
break
}
if _, err := io.ReadFull(body, header[:]); err != nil {
break
}
n := binary.BigEndian.Uint32(header[:])
if n == 0 || n > liveMaxFrame {
// A length this side cannot trust ends the stream rather than
// allocating what it was told to.
writeErr(w, http.StatusBadRequest, "bad_frame", "Frame size out of range.")
return
}
frame := make([]byte, n)
if _, err := io.ReadFull(body, frame); err != nil {
break
}
frames++
if !s.live().Publish(id, frame) {
// Nobody is watching any more. Saying so in the response is what
// stops the shop uploading; the agent goes back to waiting.
break
}
}
writeJSON(w, http.StatusOK, map[string]any{"frames": frames})
}

184
server/internal/api/live.go Normal file
View File

@@ -0,0 +1,184 @@
package api
import (
"sync"
"time"
)
// LiveHub carries camera frames from a shop PC to whoever is watching.
//
// The shop PC's engine serves MJPEG on its own loopback, behind a router with
// no inbound route, so head office cannot pull it. What head office CAN do is
// answer the agent's outbound requests - which is the whole shape of this
// product already - so the agent asks "is anyone watching?", and pushes frames
// up for as long as somebody is.
//
// This is NOT true video. It is a few frames a second of re-encoded JPEG, which
// is what an outbound HTTP relay can carry honestly. Real 25 fps video needs
// WebRTC and a TURN server; this needs neither, and for "is that camera pointed
// at the right place, and is someone in the shop" a few frames a second is what
// the question actually requires.
//
// It is deliberately the OPPOSITE of the arrivals Hub, which is a doorbell that
// pushes nothing because nothing may be lost. Here a dropped frame is the
// correct outcome: a slow viewer must never stall the pump or accumulate a
// backlog of stale pictures, because the only frame worth having is the newest
// one. So each viewer gets a one-slot buffer and a full slot is overwritten.
type LiveHub struct {
mu sync.Mutex
cameras map[string]*liveCamera
}
type liveCamera struct {
viewers map[chan []byte]struct{}
// wanted is refreshed by every watching viewer. The agent stops pushing
// when it lapses, which is what keeps a shop's uplink idle when nobody is
// looking - the entire cost argument for this feature.
wanted time.Time
// bell fires when the first viewer arrives, so an agent long-polling for
// work is answered immediately instead of on its next tick.
bell chan struct{}
}
// liveIdle is how long a camera stays "wanted" after the last viewer refreshed
// it. Longer than the viewer's refresh interval so an ordinary pause between
// refreshes does not stop the stream, short enough that a browser that
// vanished stops a shop uploading within seconds.
const liveIdle = 12 * time.Second
func NewLiveHub() *LiveHub {
return &LiveHub{cameras: map[string]*liveCamera{}}
}
// Watch registers a viewer and returns its frame channel plus a release func.
func (h *LiveHub) Watch(cameraID string) (<-chan []byte, func()) {
h.mu.Lock()
defer h.mu.Unlock()
c := h.cameras[cameraID]
if c == nil {
c = &liveCamera{viewers: map[chan []byte]struct{}{}, bell: make(chan struct{}, 1)}
h.cameras[cameraID] = c
}
ch := make(chan []byte, 1)
c.viewers[ch] = struct{}{}
c.wanted = time.Now().Add(liveIdle)
select {
case c.bell <- struct{}{}:
default:
}
return ch, func() {
h.mu.Lock()
defer h.mu.Unlock()
if cam := h.cameras[cameraID]; cam != nil {
delete(cam.viewers, ch)
if len(cam.viewers) == 0 {
// Dropped entirely rather than left empty: an estate's worth of
// cameras nobody is watching would otherwise accumulate here
// for the life of the process.
delete(h.cameras, cameraID)
}
}
close(ch)
}
}
// Keep extends a camera's interest window. Called by each viewer as it reads,
// so interest expires on its own when a browser goes away without saying so -
// which is the normal way a tab closes.
func (h *LiveHub) Keep(cameraID string) {
h.mu.Lock()
defer h.mu.Unlock()
if c := h.cameras[cameraID]; c != nil {
c.wanted = time.Now().Add(liveIdle)
}
}
// Wanted reports whether anyone is watching this camera right now.
func (h *LiveHub) Wanted(cameraID string) bool {
h.mu.Lock()
defer h.mu.Unlock()
c := h.cameras[cameraID]
return c != nil && len(c.viewers) > 0 && time.Now().Before(c.wanted)
}
// WantedAmong filters a site's cameras down to the ones being watched. The
// agent asks with the cameras it has, so this never has to know a site's
// inventory.
func (h *LiveHub) WantedAmong(ids []string) []string {
var out []string
for _, id := range ids {
if h.Wanted(id) {
out = append(out, id)
}
}
return out
}
// Bell returns a channel that fires when a viewer starts watching one of these
// cameras, so an agent waiting for work wakes at once rather than on a tick.
// A nil channel blocks forever, which is the right behaviour for the caller's
// select when none of the cameras is known here yet.
func (h *LiveHub) Bell(ids []string) <-chan struct{} {
h.mu.Lock()
defer h.mu.Unlock()
for _, id := range ids {
if c := h.cameras[id]; c != nil {
return c.bell
}
}
// Register a placeholder for the first camera so a later Watch can ring
// something. Cheap: one struct per camera an agent asked about.
if len(ids) == 0 {
return nil
}
c := &liveCamera{viewers: map[chan []byte]struct{}{}, bell: make(chan struct{}, 1)}
h.cameras[ids[0]] = c
return c.bell
}
// Publish hands one frame to every viewer of a camera and reports whether any
// remain. The agent uses that answer to stop pushing.
//
// A viewer whose slot is full has its pending frame REPLACED, never queued. The
// newest frame is the only one worth having, and a queue here would show a
// viewer an ever-growing delay behind the shop rather than dropping back to
// live.
func (h *LiveHub) Publish(cameraID string, frame []byte) bool {
h.mu.Lock()
defer h.mu.Unlock()
c := h.cameras[cameraID]
if c == nil || len(c.viewers) == 0 {
return false
}
for ch := range c.viewers {
select {
case ch <- frame:
default:
select {
case <-ch:
default:
}
select {
case ch <- frame:
default:
}
}
}
return time.Now().Before(c.wanted)
}
// live returns the hub, creating it on first use.
//
// Lazily, and stored on the Server, so a server built without one still works:
// unlike the arrivals doorbell there is no degraded mode to fall back to here,
// and a nil map panic on an endpoint somebody forgot to wire is the worst way
// to find out.
func (s *Server) live() *LiveHub {
s.liveOnce.Do(func() {
if s.Live == nil {
s.Live = NewLiveHub()
}
})
return s.Live
}

131
server/internal/api/live_test.go Normal file
View File

@@ -0,0 +1,131 @@
package api
import (
"context"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/loyaly/behavision-server/internal/auth"
)
const liveCam = "22222222-3333-4444-5555-666666666666"
// The whole cost argument for this feature: a camera nobody is watching must
// cost a shop nothing at all. If Wanted were ever optimistic, every shop PC in
// an estate would upload continuously.
func TestNobodyWatchingMeansNothingIsWanted(t *testing.T) {
h := NewLiveHub()
if h.Wanted(liveCam) {
t.Fatal("a camera nobody has asked for was reported as wanted")
}
if got := h.WantedAmong([]string{liveCam, "other"}); len(got) != 0 {
t.Fatalf("wanted %v with no viewers", got)
}
}
func TestAViewerMakesACameraWantedAndReleasingStopsIt(t *testing.T) {
h := NewLiveHub()
_, release := h.Watch(liveCam)
if !h.Wanted(liveCam) {
t.Fatal("a watched camera was not wanted")
}
release()
if h.Wanted(liveCam) {
t.Fatal("the camera stayed wanted after the last viewer left")
}
}
// The opposite policy to the arrivals Hub, and deliberately so. There nothing
// may be lost; here the only frame worth having is the newest one, and a queue
// would show a viewer an ever-growing delay behind the shop instead of dropping
// back to live.
func TestASlowViewerGetsTheNewestFrameNotTheOldest(t *testing.T) {
h := NewLiveHub()
frames, release := h.Watch(liveCam)
defer release()
for _, f := range []string{"one", "two", "three"} {
h.Publish(liveCam, []byte(f))
}
select {
case got := <-frames:
if string(got) != "three" {
t.Fatalf("a slow viewer was served %q, want the newest frame", got)
}
default:
t.Fatal("nothing was delivered")
}
}
// Publish reporting false is what stops the shop PC uploading. If it kept
// saying true the agent would push into an empty room until the session cap.
func TestPublishReportsWhenTheLastViewerHasGone(t *testing.T) {
h := NewLiveHub()
_, release := h.Watch(liveCam)
if !h.Publish(liveCam, []byte("frame")) {
t.Fatal("publish said to stop while somebody was watching")
}
release()
if h.Publish(liveCam, []byte("frame")) {
t.Fatal("publish did not say to stop after the last viewer left")
}
}
// A browser that vanishes without saying so - the normal way a tab closes -
// must stop the upload on its own.
func TestInterestExpiresWithoutBeingRefreshed(t *testing.T) {
h := NewLiveHub()
frames, release := h.Watch(liveCam)
defer release()
_ = frames
h.mu.Lock()
h.cameras[liveCam].wanted = time.Now().Add(-time.Second)
h.mu.Unlock()
if h.Wanted(liveCam) {
t.Fatal("interest did not lapse")
}
if h.Publish(liveCam, []byte("frame")) {
t.Fatal("publish kept the shop uploading for a viewer that had gone")
}
}
// The relay is keyed on a camera id and a hub does not know whose camera it is
// holding, so ownership has to be proved before anything streams. Otherwise a
// camera id - which is not a secret - would be enough to watch another
// company's shop floor.
func TestAnotherTenantCannotWatchYourCamera(t *testing.T) {
srv, fs := newServer(t)
fs.addCameraRef(liveCam, "client-1", "site-1", "cam1")
rr := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/cameras/"+liveCam+"/live", nil)
req = req.WithContext(context.WithValue(req.Context(), principalKey,
auth.Principal{ClientID: "someone-else", Role: "owner"}))
srv.handleWatchLive(rr, req)
if rr.Code != http.StatusNotFound {
t.Fatalf("status %d, want 404", rr.Code)
}
}
// An agent may only push into its OWN site's camera. The agent supplies this
// id, and a camera id is not a secret.
func TestAnAgentCannotPushIntoAnotherSitesCamera(t *testing.T) {
srv, fs := newServer(t)
fs.addCameraRef(liveCam, "client-1", "site-1", "cam1")
fs.addAgent("agent-token", AgentPrincipal{ClientID: "client-1", SiteID: "site-2"})
rr := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodPost,
"/api/agent/cameras/"+liveCam+"/live", nil)
req.Header.Set("Authorization", "Bearer agent-token")
srv.Routes().ServeHTTP(rr, req)
if rr.Code != http.StatusNotFound {
t.Fatalf("status %d, want 404", rr.Code)
}
}

5
server/internal/api/types.go
View File

@@ -440,6 +440,11 @@ type CameraInput struct {
// thing stopping a tenant response carrying camera passwords would be
// remembering to blank a field, on every path, forever.
type AgentCamera struct {
// ID is head office's uuid. Carried alongside CameraID because the two
// name the same camera to different halves of the system: head office
// addresses it by uuid, the engine on the shop PC only knows the name in
// CameraID, and the live relay has to translate between them.
ID string `json:"id"`
CameraID string `json:"camera_id"`
Label string `json:"label"`
Host string `json:"host"`