tutti/packages/agent/daemon/liveprotocol/mobile/subscriber.go

// Package liveprotocolmobile exposes the Agent live-protocol subscriber through
// gomobile without moving Agent semantics into DeviceLink.
package liveprotocolmobile

import (
	"encoding/json"
	"fmt"

	"github.com/tutti-os/tutti/packages/agent/daemon/liveprotocol"
)

type Subscriber struct {
	subscriber *liveprotocol.Subscriber
}

type applyEnvelope struct {
	Accepted          []deliveryEnvelope `json:"accepted"`
	AfterSeq          int64              `json:"afterSeq"`
	DuplicateCount    int                `json:"duplicateCount"`
	Epoch             int64              `json:"epoch"`
	Reason            string             `json:"reason,omitempty"`
	ReconcileRequired bool               `json:"reconcileRequired"`
}

type deliveryEnvelope struct {
	Kind               string                           `json:"kind"`
	Seq                int64                            `json:"seq"`
	Event              json.RawMessage                  `json:"event,omitempty"`
	Discontinuity      *liveprotocol.Discontinuity      `json:"discontinuity,omitempty"`
	AttachmentChanged  *liveprotocol.AttachmentChanged  `json:"attachmentChanged,omitempty"`
	AttachmentCaughtUp *liveprotocol.AttachmentCaughtUp `json:"attachmentCaughtUp,omitempty"`
	GoalChanged        *liveprotocol.GoalChanged        `json:"goalChanged,omitempty"`
	StreamReady        *liveprotocol.StreamReady        `json:"streamReady,omitempty"`
	Rejected           *liveprotocol.Rejected           `json:"rejected,omitempty"`
}

func ProtocolRevision() string {
	return liveprotocol.ProtocolRevision
}

func NewSubscriber(epoch int64, afterSeq int64) (*Subscriber, error) {
	if epoch < 0 || afterSeq < 0 {
		return nil, fmt.Errorf("agent live resume cursor must not be negative")
	}
	subscriber, err := liveprotocol.NewSubscriber(liveprotocol.SubscriberConfig{
		Epoch:    uint64(epoch),
		AfterSeq: uint64(afterSeq),
	})
	if err != nil {
		return nil, err
	}
	return &Subscriber{subscriber: subscriber}, nil
}

// Apply validates one protobuf-wire frame and returns only accepted,
// contiguous deliveries as JSON for the React Native boundary.
func (s *Subscriber) Apply(encoded []byte) (string, error) {
	if s == nil || s.subscriber == nil {
		return "", fmt.Errorf("agent live subscriber is unavailable")
	}
	result, err := liveprotocol.DecodeAndApply(s.subscriber, encoded)
	if err != nil {
		return "", err
	}
	cursor := s.subscriber.ResumeCursor()
	if cursor.Epoch > uint64(^uint64(0)>>1) || cursor.AfterSeq > uint64(^uint64(0)>>1) {
		return "", fmt.Errorf("agent live cursor exceeds mobile integer range")
	}
	envelope := applyEnvelope{
		Accepted:          make([]deliveryEnvelope, 0, len(result.Accepted)),
		AfterSeq:          int64(cursor.AfterSeq),
		DuplicateCount:    result.DuplicateCount,
		Epoch:             int64(cursor.Epoch),
		Reason:            result.Reason,
		ReconcileRequired: result.ReconcileRequired,
	}
	for _, delivery := range result.Accepted {
		envelope.Accepted = append(envelope.Accepted, deliveryEnvelope{
			Kind:               deliveryKindName(delivery.Kind),
			Seq:                int64(delivery.Seq),
			Event:              append(json.RawMessage(nil), delivery.Event...),
			Discontinuity:      delivery.Discontinuity,
			AttachmentChanged:  delivery.AttachmentChanged,
			AttachmentCaughtUp: delivery.AttachmentCaughtUp,
			GoalChanged:        delivery.GoalChanged,
			StreamReady:        delivery.StreamReady,
			Rejected:           delivery.Rejected,
		})
	}
	raw, err := json.Marshal(envelope)
	if err != nil {
		return "", fmt.Errorf("encode agent live mobile delivery: %w", err)
	}
	return string(raw), nil
}

func deliveryKindName(kind liveprotocol.DeliveryKind) string {
	switch kind {
	case liveprotocol.DeliveryKindEvent:
		return "event"
	case liveprotocol.DeliveryKindDiscontinuity:
		return "discontinuity"
	case liveprotocol.DeliveryKindAttachmentChanged:
		return "attachment_changed"
	case liveprotocol.DeliveryKindAttachmentCaughtUp:
		return "attachment_caught_up"
	case liveprotocol.DeliveryKindGoalChanged:
		return "goal_changed"
	case liveprotocol.DeliveryKindStreamReady:
		return "stream_ready"
	case liveprotocol.DeliveryKindRejected:
		return "rejected"
	default:
		return "unknown"
	}
}