teraWeb/backend/internal/serial/stream.go

package serial

import (
	"context"
	"crypto/rand"
	"encoding/base64"
	"errors"
	"fmt"
	"io"
	"regexp"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

const (
	DefaultStreamSocketPath = "/run/tera-seriald/serial-stream.sock"
	MaxCapturedPayloadBytes = 4 << 10
	CRProbeClientID         = "tera-serial-link-probe"
)

var (
	ErrStreamLeaseBusy  = errors.New("serial stream lease is already active")
	streamClientPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$`)
)

// StreamTarget is a short-lived reservation of one validated mapping entry.
// Release is idempotent and must be called if opening the device fails.
type StreamTarget struct {
	Channel string
	Device  string
	Group   string

	releaseOnce sync.Once
	release     func()
}

func newStreamTarget(channel, device, group string, release func()) *StreamTarget {
	return &StreamTarget{Channel: channel, Device: device, Group: group, release: release}
}

func (t *StreamTarget) Release() {
	if t == nil {
		return
	}
	t.releaseOnce.Do(func() {
		if t.release != nil {
			t.release()
		}
	})
}

// StreamResolver prevents a stream lease from bypassing the board-validated
// CH mapping. MappingController is the production implementation.
type StreamResolver interface {
	ReserveStream(context.Context, string) (*StreamTarget, error)
}

type StreamOpener interface {
	OpenStream(path string) (io.ReadWriteCloser, error)
}

type platformStreamOpener struct{}

func (platformStreamOpener) OpenStream(path string) (io.ReadWriteCloser, error) {
	return openSerialDevice(path)
}

type StreamEvent struct {
	Timestamp     time.Time `json:"timestamp"`
	Channel       string    `json:"channel"`
	LeaseID       string    `json:"lease_id"`
	ClientID      string    `json:"client_id"`
	Direction     string    `json:"direction"`
	Bytes         uint64    `json:"bytes,omitempty"`
	Payload       []byte    `json:"payload,omitempty"`
	LogFormat     string    `json:"log_format,omitempty"`
	ShowTimestamp bool      `json:"show_timestamp"`
	ShowDirection bool      `json:"show_direction"`
	ShowUsername  bool      `json:"show_username"`
	ShowIPAddress bool      `json:"show_ip_address"`
	ShowPort      bool      `json:"show_port"`
}

// StreamEventSink is deliberately non-blocking. Payload ownership transfers
// to the sink; BoundedStreamEventSink copies it before queueing so caller
// buffers can be reused immediately.
type StreamEventSink interface {
	TryRecord(StreamEvent) bool
}

type BoundedStreamEventSink struct {
	events  chan StreamEvent
	dropped atomic.Uint64
}

func NewBoundedStreamEventSink(capacity int) *BoundedStreamEventSink {
	if capacity < 1 {
		capacity = 1
	}
	if capacity > 4096 {
		capacity = 4096
	}
	return &BoundedStreamEventSink{events: make(chan StreamEvent, capacity)}
}

func (s *BoundedStreamEventSink) TryRecord(event StreamEvent) bool {
	if s == nil {
		return true
	}
	if len(event.Payload) > 0 {
		event.Payload = append([]byte(nil), event.Payload...)
	}
	select {
	case s.events <- event:
		return true
	default:
		s.dropped.Add(1)
		return false
	}
}

func (s *BoundedStreamEventSink) Events() <-chan StreamEvent { return s.events }
func (s *BoundedStreamEventSink) Dropped() uint64            { return s.dropped.Load() }

type StreamLeaseOptions struct {
	Opener StreamOpener
	Events StreamEventSink
}

type StreamSnapshot struct {
	ActiveLeases  int                     `json:"active_leases"`
	RXBytes       uint64                  `json:"rx_bytes"`
	TXBytes       uint64                  `json:"tx_bytes"`
	DroppedEvents uint64                  `json:"dropped_events"`
	Channels      []StreamChannelSnapshot `json:"channels,omitempty"`
}

type StreamChannelSnapshot struct {
	Channel  string     `json:"channel"`
	Active   bool       `json:"active"`
	RXBytes  uint64     `json:"rx_bytes"`
	TXBytes  uint64     `json:"tx_bytes"`
	LastRXAt *time.Time `json:"last_rx_at,omitempty"`
}

type streamChannelCounters struct {
	rx                   atomic.Uint64
	tx                   atomic.Uint64
	lastRXAt             atomic.Int64
	lastExternalActivity atomic.Int64
}

type StreamLeaseManager struct {
	resolver StreamResolver
	opener   StreamOpener
	events   StreamEventSink
	capture  map[string]CaptureConfig
	buffers  map[string]*captureAccumulator

	mu        sync.Mutex
	captureMu sync.Mutex
	openMu    sync.Mutex
	openLocks map[string]*sync.Mutex
	active    map[string]*StreamLease
	counts    map[string]*streamChannelCounters
	rx        atomic.Uint64
	tx        atomic.Uint64
	drops     atomic.Uint64
}

type captureAccumulator struct {
	channel   string
	leaseID   string
	clientID  string
	direction string
	config    CaptureConfig
	payload   []byte
	timer     *time.Timer
}

func NewStreamLeaseManager(resolver StreamResolver, options StreamLeaseOptions) (*StreamLeaseManager, error) {
	if resolver == nil {
		return nil, errors.New("serial stream resolver is required")
	}
	opener := options.Opener
	if opener == nil {
		opener = platformStreamOpener{}
	}
	manager := &StreamLeaseManager{
		resolver:  resolver,
		opener:    opener,
		events:    options.Events,
		active:    make(map[string]*StreamLease),
		capture:   defaultCaptureModes(),
		buffers:   make(map[string]*captureAccumulator),
		counts:    make(map[string]*streamChannelCounters, ChannelCount),
		openLocks: make(map[string]*sync.Mutex),
	}
	for index := 1; index <= ChannelCount; index++ {
		manager.counts[fmt.Sprintf("CH%02d", index)] = &streamChannelCounters{}
	}
	return manager, nil
}

// ApplyCapture atomically changes payload logging without touching an active
// UART lease. Unspecified channels keep their current setting.
func (m *StreamLeaseManager) ApplyCapture(ctx context.Context, configs []CaptureConfig) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	normalized, err := normalizeCaptureConfigs(configs)
	if err != nil {
		return err
	}
	m.mu.Lock()
	for channel, config := range normalized {
		m.capture[channel] = config
	}
	m.mu.Unlock()
	m.captureMu.Lock()
	for key, accumulator := range m.buffers {
		if _, changed := normalized[accumulator.channel]; !changed {
			continue
		}
		if accumulator.timer != nil {
			accumulator.timer.Stop()
		}
		delete(m.buffers, key)
	}
	m.captureMu.Unlock()
	return nil
}

func (m *StreamLeaseManager) Acquire(ctx context.Context, channel, clientID string) (*StreamLease, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	canonical, err := normalizeChannel(channel)
	if err != nil || canonical != channel {
		return nil, errors.New("stream channel must be canonical CH01 through CH48")
	}
	if !streamClientPattern.MatchString(clientID) {
		return nil, errors.New("stream client_id is invalid")
	}

	m.mu.Lock()
	if _, exists := m.active[channel]; exists {
		m.mu.Unlock()
		return nil, ErrStreamLeaseBusy
	}
	// A nil entry reserves the channel while mapping validation and open run.
	m.active[channel] = nil
	m.mu.Unlock()

	failed := true
	defer func() {
		if failed {
			m.mu.Lock()
			if current, exists := m.active[channel]; exists && current == nil {
				delete(m.active, channel)
			}
			m.mu.Unlock()
		}
	}()

	target, err := m.resolver.ReserveStream(ctx, channel)
	if err != nil {
		return nil, err
	}
	if target == nil || target.Channel != channel || target.Device == "" {
		if target != nil {
			target.Release()
		}
		return nil, &UnavailableError{Reason: "serial stream resolver returned an invalid target"}
	}
	if err := ctx.Err(); err != nil {
		target.Release()
		return nil, err
	}
	openLock := m.openLock(target.Group, target.Device)
	openLock.Lock()
	device, err := m.opener.OpenStream(target.Device)
	openLock.Unlock()
	if err != nil {
		target.Release()
		return nil, &UnavailableError{Reason: fmt.Sprintf("%s serial stream device cannot be opened", channel)}
	}
	leaseID, err := newStreamLeaseID()
	if err != nil {
		_ = device.Close()
		target.Release()
		return nil, err
	}
	lease := &StreamLease{
		manager: m, target: target, device: device,
		channel: channel, clientID: clientID, leaseID: leaseID,
	}
	m.mu.Lock()
	m.active[channel] = lease
	m.mu.Unlock()
	failed = false
	if clientID != CRProbeClientID {
		m.emit(StreamEvent{Channel: channel, LeaseID: leaseID, ClientID: clientID, Direction: "open"})
	}
	return lease, nil
}

// openLock serializes the short open(2) path for UARTs sharing one PCI
// function/IRQ. Active leases remain independent after the device is open.
func (m *StreamLeaseManager) openLock(group, device string) *sync.Mutex {
	key := strings.TrimSpace(group)
	if key == "" {
		key = device
	}
	m.openMu.Lock()
	defer m.openMu.Unlock()
	lock := m.openLocks[key]
	if lock == nil {
		lock = &sync.Mutex{}
		m.openLocks[key] = lock
	}
	return lock
}

func (m *StreamLeaseManager) Snapshot() StreamSnapshot {
	m.mu.Lock()
	active := 0
	activeChannels := make(map[string]bool, len(m.active))
	for channel, lease := range m.active {
		if lease != nil && lease.clientID == CRProbeClientID {
			continue
		}
		active++
		activeChannels[channel] = true
	}
	m.mu.Unlock()
	channels := make([]StreamChannelSnapshot, 0, len(m.counts))
	for index := 1; index <= ChannelCount; index++ {
		channel := fmt.Sprintf("CH%02d", index)
		counters := m.counts[channel]
		rxBytes := counters.rx.Load()
		txBytes := counters.tx.Load()
		lastRXNanoseconds := counters.lastRXAt.Load()
		if !activeChannels[channel] && rxBytes == 0 && txBytes == 0 {
			continue
		}
		var lastRXAt *time.Time
		if lastRXNanoseconds > 0 {
			value := time.Unix(0, lastRXNanoseconds).UTC()
			lastRXAt = &value
		}
		channels = append(channels, StreamChannelSnapshot{
			Channel: channel, Active: activeChannels[channel], RXBytes: rxBytes, TXBytes: txBytes, LastRXAt: lastRXAt,
		})
	}
	return StreamSnapshot{
		ActiveLeases: active,
		RXBytes:      m.rx.Load(), TXBytes: m.tx.Load(), DroppedEvents: m.drops.Load(),
		Channels: channels,
	}
}

// RecentlyActive reports user/session traffic only. Internal CR probes do not
// extend the quiet period and are not exposed as active user sessions.
func (m *StreamLeaseManager) RecentlyActive(channel string, quietPeriod time.Duration, now time.Time) bool {
	if m == nil || quietPeriod <= 0 {
		return false
	}
	counters, ok := m.counts[channel]
	if !ok {
		return false
	}
	nanoseconds := counters.lastExternalActivity.Load()
	if nanoseconds <= 0 {
		return false
	}
	return now.Sub(time.Unix(0, nanoseconds)) < quietPeriod
}

func (m *StreamLeaseManager) Close() error {
	m.mu.Lock()
	leases := make([]*StreamLease, 0, len(m.active))
	for _, lease := range m.active {
		if lease != nil {
			leases = append(leases, lease)
		}
	}
	m.mu.Unlock()
	var closeErr error
	for _, lease := range leases {
		if err := lease.Close(); err != nil {
			closeErr = errors.Join(closeErr, err)
		}
	}
	return closeErr
}

func (m *StreamLeaseManager) release(lease *StreamLease) {
	m.flushCaptureLease(lease.leaseID)
	m.mu.Lock()
	if current, exists := m.active[lease.channel]; exists && current == lease {
		delete(m.active, lease.channel)
	}
	m.mu.Unlock()
}

func (m *StreamLeaseManager) emit(event StreamEvent) {
	if event.Timestamp.IsZero() {
		event.Timestamp = time.Now().UTC()
	}
	if m.events != nil && !m.events.TryRecord(event) {
		m.drops.Add(1)
	}
}

func (m *StreamLeaseManager) emitPayload(channel, leaseID, clientID, direction string, payload []byte) {
	config, enabled := m.captureConfig(channel, direction)
	if len(payload) == 0 || !enabled {
		return
	}
	trigger := config.DataLog.ReadTrigger
	if direction == CaptureModeTX {
		trigger = config.DataLog.SendTrigger
	}
	if config.DataPackage.Mode == "disabled" {
		for len(payload) > 0 {
			size := min(len(payload), MaxCapturedPayloadBytes)
			m.emitCaptured(channel, leaseID, clientID, direction, config, trigger, payload[:size])
			payload = payload[size:]
		}
		return
	}
	key := leaseID + "\x00" + direction
	var ready [][]byte
	m.captureMu.Lock()
	accumulator := m.buffers[key]
	if accumulator == nil {
		accumulator = &captureAccumulator{channel: channel, leaseID: leaseID, clientID: clientID, direction: direction, config: config}
		m.buffers[key] = accumulator
	}
	if accumulator.timer != nil {
		accumulator.timer.Stop()
		accumulator.timer = nil
	}
	accumulator.config = config
	accumulator.payload = append(accumulator.payload, payload...)
	for len(accumulator.payload) > 0 {
		boundary := captureFrameBoundary(accumulator.payload, config, trigger)
		if boundary == 0 {
			break
		}
		ready = append(ready, append([]byte(nil), accumulator.payload[:boundary]...))
		accumulator.payload = accumulator.payload[boundary:]
	}
	if len(accumulator.payload) > 0 {
		if timeout := captureFlushTimeout(config, trigger); timeout > 0 {
			accumulator.timer = time.AfterFunc(timeout, func() { m.flushCaptureKey(key) })
		} else if captureFrameBoundary(accumulator.payload, config, trigger) == 0 {
			ready = append(ready, append([]byte(nil), accumulator.payload...))
			accumulator.payload = nil
		}
	}
	if len(accumulator.payload) == 0 {
		delete(m.buffers, key)
	}
	m.captureMu.Unlock()
	for _, frame := range ready {
		m.emitCaptured(channel, leaseID, clientID, direction, config, trigger, frame)
	}
}

func (m *StreamLeaseManager) emitCaptured(channel, leaseID, clientID, direction string, config CaptureConfig, trigger CaptureTriggerConfig, payload []byte) {
	chunk := append([]byte(nil), payload...)
	chunk = processCaptureDelimiter(chunk, trigger.Process, trigger.Character1, trigger.Character2)
	chunk = processCaptureDelimiter(chunk, config.DataPackage.DelimiterProcess, config.DataPackage.Delimiter1, config.DataPackage.Delimiter2)
	if len(chunk) == 0 {
		return
	}
	m.emit(StreamEvent{
		Channel: channel, LeaseID: leaseID, ClientID: clientID,
		Direction: direction, Bytes: uint64(len(chunk)), Payload: chunk,
		LogFormat: config.DataLog.Format, ShowTimestamp: config.DataLog.ShowTimestamp,
		ShowDirection: config.DataLog.ShowDirection, ShowUsername: config.DataLog.ShowUsername,
		ShowIPAddress: config.DataLog.ShowIPAddress, ShowPort: config.DataLog.ShowPort,
	})
}

func (m *StreamLeaseManager) flushCaptureKey(key string) {
	m.captureMu.Lock()
	accumulator := m.buffers[key]
	if accumulator == nil {
		m.captureMu.Unlock()
		return
	}
	delete(m.buffers, key)
	if accumulator.timer != nil {
		accumulator.timer.Stop()
	}
	payload := append([]byte(nil), accumulator.payload...)
	m.captureMu.Unlock()
	trigger := accumulator.config.DataLog.ReadTrigger
	if accumulator.direction == CaptureModeTX {
		trigger = accumulator.config.DataLog.SendTrigger
	}
	m.emitCaptured(accumulator.channel, accumulator.leaseID, accumulator.clientID, accumulator.direction, accumulator.config, trigger, payload)
}

func (m *StreamLeaseManager) flushCaptureLease(leaseID string) {
	prefix := leaseID + "\x00"
	m.captureMu.Lock()
	keys := make([]string, 0, 2)
	for key := range m.buffers {
		if strings.HasPrefix(key, prefix) {
			keys = append(keys, key)
		}
	}
	m.captureMu.Unlock()
	for _, key := range keys {
		m.flushCaptureKey(key)
	}
}

func captureFrameBoundary(payload []byte, config CaptureConfig, trigger CaptureTriggerConfig) int {
	boundary := 0
	consider := func(candidate int) {
		if candidate > 0 && (boundary == 0 || candidate < boundary) {
			boundary = candidate
		}
	}
	consider(min(len(payload), MaxCapturedPayloadBytes))
	if len(payload) < MaxCapturedPayloadBytes {
		boundary = 0
	}
	if (config.DataPackage.Mode == "length" || config.DataPackage.Mode == "interval") && config.DataPackage.Length > 0 && len(payload) >= config.DataPackage.Length {
		consider(config.DataPackage.Length)
	}
	if config.DataPackage.Mode == "delimiter" {
		consider(captureDelimiterBoundary(payload, config.DataPackage.Delimiter1, config.DataPackage.Delimiter2))
	}
	if trigger.Length > 0 && len(payload) >= trigger.Length {
		consider(trigger.Length)
	}
	consider(captureDelimiterBoundary(payload, trigger.Character1, trigger.Character2))
	return boundary
}

func captureDelimiterBoundary(payload []byte, first, second int) int {
	if first < 0 && second < 0 {
		return 0
	}
	if first >= 0 && second >= 0 {
		for index := 0; index+1 < len(payload); index++ {
			if int(payload[index]) == first && int(payload[index+1]) == second {
				return index + 2
			}
		}
		return 0
	}
	expected := first
	if expected < 0 {
		expected = second
	}
	for index, value := range payload {
		if int(value) == expected {
			return index + 1
		}
	}
	return 0
}

func captureFlushTimeout(config CaptureConfig, trigger CaptureTriggerConfig) time.Duration {
	milliseconds := 0
	consider := func(value int) {
		if value > 0 && (milliseconds == 0 || value < milliseconds) {
			milliseconds = value
		}
	}
	if config.DataPackage.Mode == "interval" || config.DataPackage.Mode == "timeout" || config.DataPackage.Mode == "delimiter" {
		consider(config.DataPackage.TimeoutMillis)
	}
	consider(trigger.TimeoutMillis)
	return time.Duration(milliseconds) * time.Millisecond
}

func (m *StreamLeaseManager) captureConfig(channel, direction string) (CaptureConfig, bool) {
	m.mu.Lock()
	config := m.capture[channel]
	m.mu.Unlock()
	enabled := config.Mode == CaptureModeRXTX || config.Mode == direction
	if direction == CaptureModeRX {
		enabled = enabled && config.DataLog.Read
	} else if direction == CaptureModeTX {
		enabled = enabled && config.DataLog.Send
	}
	return config, enabled
}

func defaultCaptureModes() map[string]CaptureConfig {
	modes := make(map[string]CaptureConfig, ChannelCount)
	for index := 1; index <= ChannelCount; index++ {
		channel := fmt.Sprintf("CH%02d", index)
		modes[channel] = CaptureConfig{
			Channel: channel, Mode: CaptureModeRXTX,
			DataPackage: CapturePackageConfig{Mode: "interval", Length: MaxCapturedPayloadBytes, DelimiterProcess: "keep", Delimiter1: -1, Delimiter2: -1},
			DataLog: CaptureLogConfig{
				Read: true, Send: true, ShowTimestamp: true, ShowDirection: true, Format: "ascii",
				ReadTrigger: CaptureTriggerConfig{Process: "keep", Character1: -1, Character2: -1},
				SendTrigger: CaptureTriggerConfig{Process: "keep", Character1: -1, Character2: -1},
			},
		}
	}
	return modes
}

func processCaptureDelimiter(payload []byte, process string, first, second int) []byte {
	if strings.ToLower(strings.TrimSpace(process)) != "remove" {
		return payload
	}
	result := payload[:0]
	for _, value := range payload {
		if (first >= 0 && int(value) == first) || (second >= 0 && int(value) == second) {
			continue
		}
		result = append(result, value)
	}
	return result
}

type StreamLease struct {
	manager  *StreamLeaseManager
	target   *StreamTarget
	device   io.ReadWriteCloser
	channel  string
	clientID string
	leaseID  string
	writeMu  sync.Mutex

	closeOnce sync.Once
	closeErr  error
}

func (l *StreamLease) LeaseID() string { return l.leaseID }

func (l *StreamLease) Read(buffer []byte) (int, error) {
	n, err := l.device.Read(buffer)
	if n > 0 && l.clientID != CRProbeClientID {
		now := time.Now()
		l.manager.rx.Add(uint64(n))
		l.manager.counts[l.channel].rx.Add(uint64(n))
		l.manager.counts[l.channel].lastRXAt.Store(now.UnixNano())
		l.manager.counts[l.channel].lastExternalActivity.Store(now.UnixNano())
		l.manager.emitPayload(l.channel, l.leaseID, l.clientID, CaptureModeRX, buffer[:n])
	}
	return n, err
}

func (l *StreamLease) Write(buffer []byte) (int, error) {
	l.writeMu.Lock()
	defer l.writeMu.Unlock()
	n, err := l.device.Write(buffer)
	if n > 0 && l.clientID != CRProbeClientID {
		l.manager.tx.Add(uint64(n))
		l.manager.counts[l.channel].tx.Add(uint64(n))
		l.manager.counts[l.channel].lastExternalActivity.Store(time.Now().UnixNano())
		l.manager.emitPayload(l.channel, l.leaseID, l.clientID, CaptureModeTX, buffer[:n])
	}
	return n, err
}

// SendBreak emits a real UART break condition through tera-seriald. It is
// serialized with ordinary TX so terminal data cannot overlap the ioctl.
func (l *StreamLease) SendBreak(ctx context.Context) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	l.writeMu.Lock()
	defer l.writeMu.Unlock()
	if sender, ok := l.device.(interface{ SendBreak() error }); ok {
		return sender.SendBreak()
	}
	return platformSendSerialBreak(l.device)
}

func (m *StreamLeaseManager) SendBreak(ctx context.Context, channel, leaseID, clientID string) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	m.mu.Lock()
	lease := m.active[channel]
	m.mu.Unlock()
	if lease == nil || lease.leaseID != leaseID || lease.clientID != clientID {
		return errors.New("serial stream break lease is not active")
	}
	return lease.SendBreak(ctx)
}

func (l *StreamLease) Close() error {
	l.closeOnce.Do(func() {
		l.closeErr = l.device.Close()
		l.target.Release()
		l.manager.release(l)
		if l.clientID != CRProbeClientID {
			l.manager.emit(StreamEvent{
				Channel: l.channel, LeaseID: l.leaseID, ClientID: l.clientID, Direction: "close",
			})
		}
	})
	return l.closeErr
}

func newStreamLeaseID() (string, error) {
	buffer := make([]byte, 16)
	if _, err := io.ReadFull(rand.Reader, buffer); err != nil {
		return "", fmt.Errorf("generate serial stream lease id: %w", err)
	}
	return base64.RawURLEncoding.EncodeToString(buffer), nil
}

// ReserveStream is the only production path from a canonical CH label to a
// UART stream device. It also blocks line-setting mutation for the lease.
func (c *MappingController) ReserveStream(ctx context.Context, channel string) (*StreamTarget, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	canonical, err := normalizeChannel(channel)
	if err != nil || canonical != channel {
		return nil, errors.New("stream channel must be canonical CH01 through CH48")
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	if _, exists := c.activeStreams[channel]; exists {
		return nil, ErrStreamLeaseBusy
	}
	entry, ok := c.byChannel[channel]
	if !ok {
		return nil, &UnavailableError{Reason: "channel is not in the validated serial mapping"}
	}
	state := c.inspectState(entry)
	if !state.Available {
		return nil, &UnavailableError{Reason: channel + " " + state.Message}
	}
	var target *StreamTarget
	target = newStreamTarget(channel, entry.Device, entry.PCIBDF, func() {
		c.mu.Lock()
		defer c.mu.Unlock()
		if current, exists := c.activeStreams[channel]; exists && current == target {
			delete(c.activeStreams, channel)
		}
	})
	c.activeStreams[channel] = target
	return target, nil
}