Set up your own workflow stage

mkdir hub-loitering && cd hub-loitering
go mod init github.com/uug-ai/hub-loitering
go get github.com/uug-ai/models@v1.6.3
go get github.com/uug-ai/queue@v1.3.6
go get github.com/sirupsen/logrus@v1.9.4

package main

import (
	"encoding/json"
	"os"
	"strings"
	"time"

	"github.com/sirupsen/logrus"

	"github.com/uug-ai/models/pkg/models"
	queue "github.com/uug-ai/queue/pkg/queue"
)

// operation is this stage's id. It must match the operation registered in the
// Helm chart, and it is the key the engine files your result under.
const operation = "loitering"

func main() {
	logger := logrus.New()
	logger.SetFormatter(&logrus.JSONFormatter{})

	// The stage queue we consume from, and the engine queue we return to.
	stageQueue := envOr("LOITERING_QUEUE", "hub-workflows-loitering")
	workflowsQueue := envOr("WORKFLOWS_QUEUE", "hub-workflows-queue")

	// A stage only consumes and dead-letters; it never forwards down a stage
	// list, so it sets just its consume queue and a deadletter queue.
	options := queue.NewRabbitOptions().
		SetHost(os.Getenv("RABBITMQ_HOST")).
		SetExchange(os.Getenv("RABBITMQ_EXCHANGE")).
		SetUsername(os.Getenv("RABBITMQ_USERNAME")).
		SetPassword(os.Getenv("RABBITMQ_PASSWORD")).
		SetWorkflowsStageQueue(stageQueue).
		SetDeadletterQueue("dead-letter-queue").
		Build()

	q, err := queue.New(options)
	if err != nil {
		logger.Fatalf("failed to create queue: %v", err)
	}
	if err := q.Client.Connect(); err != nil {
		logger.Fatalf("failed to connect to broker: %v", err)
	}
	logger.Infof("loitering started: consuming %q, returning results to %q", stageQueue, workflowsQueue)

	// The workflow subsystem exchanges models.WorkflowRun (not the pipeline's
	// PipelineEvent), so decode the run ourselves. A body that isn't a
	// WorkflowRun is dead-lettered; otherwise the handler routes it back.
	handler := func(payload []byte, _ ...any) (models.PipelineAction, []byte, int) {
		var run models.WorkflowRun
		if err := json.Unmarshal(payload, &run); err != nil {
			logger.Errorf("not a WorkflowRun, dead-lettering: %v", err)
			return models.PipelineError, payload, 0
		}
		return handleRun(logger, q.Client, workflowsQueue, &run), payload, 0
	}

	rmq, ok := q.Client.(*queue.RabbitMQ)
	if !ok {
		logger.Fatalf("loitering requires a *queue.RabbitMQ client, got %T", q.Client)
	}
	for {
		if err := rmq.ReadRawMessages(handler, func(models.PipelineMetrics) {}); err != nil {
			logger.Errorf("failed to read messages: %v", err)
		}
		if err := rmq.Reconnect(); err != nil {
			logger.Errorf("failed to reconnect: %v", err)
			time.Sleep(5 * time.Second)
		}
	}
}

func envOr(name, fallback string) string {
	if v := strings.TrimSpace(os.Getenv(name)); v != "" {
		return v
	}
	return fallback
}

package main

import (
	"github.com/uug-ai/models/pkg/models"
)

// measureLoitering is where your stage's logic runs. Here it reads the dwell
// span the classifier already tracked and turns the longest one into a marker.
// A heavier stage would instead fetch the clip referenced by run.Key with the
// credentials in run.Storage and run its own model. This stub returns one fixed
// 30-second marker so you can prove the wiring before plugging in real analysis.
func measureLoitering(run *models.WorkflowRun) models.Marker {
	const start, duration = 1_700_000_000, 30 // recording epoch + dwell seconds
	return models.Marker{
		Name:           "loitering-person-1",
		Description:    "Subject lingered in frame",
		StartTimestamp: start,
		EndTimestamp:   start + duration,
		Categories:     []models.MarkerCategory{{Name: "alert"}, {Name: "person"}},
		Tags:           []models.MarkerTag{{Name: operation}},
		Events: []models.MarkerEvent{{
			Name:           "Loitering",
			StartTimestamp: start,
			EndTimestamp:   start + duration,
		}},
	}
}

{
  "blocks": [
    {
      "type": "marker",
      "data": { "name": "loitering-person-1", "startTimestamp": 1700000000, "endTimestamp": 1700000030 }
    }
  ]
}

import (
	"encoding/json"

	"github.com/sirupsen/logrus"

	ingest "github.com/uug-ai/ingest/pkg/ingest"
	queue "github.com/uug-ai/queue/pkg/queue"
)

func handleRun(logger *logrus.Logger, q queue.QueueInterface, workflowsQueue string, run *models.WorkflowRun) models.PipelineAction {
	logger.WithFields(logrus.Fields{
		"operation": run.Operation,
		"runId":     run.RunId,
		"mediaKey":  run.Key,
		"deviceKey": run.Device.DeviceKey,
	}).Info("loitering received dispatch")

	// 1. Measure the dwell (Step 3).
	marker := measureLoitering(run)

	// 2. Wrap the marker in a block envelope. A loitering stage emits one marker
	//    block; a richer stage could append detections or other block types in
	//    the same list. The platform keys the marker by (device, name,
	//    startTimestamp), so a redelivery refreshes it instead of duplicating.
	data, err := json.Marshal(marker)
	if err != nil {
		logger.Errorf("failed to marshal marker: %v", err)
		return models.PipelineCancel
	}
	envelope, err := json.Marshal(ingest.BlockEnvelope{
		Blocks: []ingest.Block{{Type: ingest.KindMarker, Data: data}},
	})
	if err != nil {
		logger.Errorf("failed to marshal envelope: %v", err)
		return models.PipelineCancel
	}

	// 3. Return the SAME run: echo identity, clear storage, attach the envelope.
	result := *run
	result.Operation = operation
	result.Storage = nil // credentials are never echoed back
	result.Payload = envelope

	body, err := json.Marshal(&result)
	if err != nil {
		logger.Errorf("failed to marshal result: %v", err)
		return models.PipelineCancel
	}
	if err := q.Publish(workflowsQueue, body); err != nil {
		logger.Errorf("failed to return result to %q: %v", workflowsQueue, err)
	}

	// We've already routed the result back ourselves, so there's nothing for the
	// queue library to forward.
	return models.PipelineCancel
}

go mod tidy
go build ./...

# Dockerfile
FROM golang:1.25-bookworm AS builder
WORKDIR /src
COPY go.mod go.sum ./
RUN go mod download
COPY . .
RUN CGO_ENABLED=0 go build -tags timetzdata,netgo -ldflags '-s -w' -o /out/loitering .

FROM alpine:latest
RUN apk add --no-cache ca-certificates && adduser -S loitering
USER loitering
COPY --from=builder /out/loitering /usr/local/bin/loitering
ENTRYPOINT ["loitering"]

docker build -t ghcr.io/uug-ai/hub-loitering:v1.0.0 .
docker push ghcr.io/uug-ai/hub-loitering:v1.0.0

kerberoshub:
  workflows:
    enabled: true                  # master switch: the workflows engine

    # ── the workflow stage object: declare + route ──────────────
    stages:
      loitering:                   # stage id (and default operation id)
        enabled: true              # route to this stage
        dispatch: conditional      # only on matching recordings (see below)
        needsMode: any
        needs:
          - operation: classify    # wait until the classifier has run…
            condition:             # …and only if it saw a person
              path: inputs.classify.properties
              op: contains
              value: person

  # ── the deployments: your microservice ────────────────────────────
  services:
    loitering:                     # same name as the stage object
      enabled: true                # deploy the microservice pod
      repository: ghcr.io/uug-ai/hub-loitering
      tag: "v1.0.0"
      queue: "hub-workflows-loitering" # the queue your microservice consumes
      replicas: 1
      pullPolicy: IfNotPresent
      logLevel: info

helm upgrade hub kerberos/hub -n kerberos-hub -f values.yaml

kubectl -n kerberos-hub get pods | grep -E 'workflows|loitering'
kubectl -n kerberos-hub logs deploy/hub-loitering
   # loitering started: consuming "hub-workflows-loitering", returning results to "hub-workflows-queue"