mxaccessgw/clients/go/mxgateway/session.go

package mxgateway

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"sort"
	"sync"
	"time"

	pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

const maxBulkItems = 1000

// eventBufferSize is the number of buffered event slots on the
// Events/EventsAfter (cancel-when-full) results channel.
const eventBufferSize = 16

// eventBufferReservedSlots is the extra capacity reserved beyond
// eventBufferSize so a terminal ErrSlowConsumer result can always be enqueued
// non-blockingly on overflow, even when all data slots are full.
const eventBufferReservedSlots = 1

// EventResult carries the next ordered event, a replay-gap signal, or a
// terminal stream error. Exactly one of Event, ReplayGap, or Err is set on any
// delivered result.
type EventResult struct {
	// Event is the next MXAccess event from the stream when both ReplayGap and
	// Err are nil.
	Event *MxEvent
	// ReplayGap, when non-nil, is the gateway's reconnect-replay gap sentinel: it
	// is delivered at the head of a resumed stream (one opened with a non-zero
	// after_worker_sequence via EventsAfter/SubscribeEventsAfter) when the
	// requested sequence predates the oldest event still retained in the replay
	// ring, so the events in between were lost.
	//
	// It is a non-terminal, observable signal — the stream continues with normal
	// events after it, and Event is nil on a gap result so a gap is never
	// mistaken for a normal MXAccess event. On seeing a gap the consumer must
	// discard any locally cached tag/alarm state and re-snapshot. To resume
	// cleanly (without provoking another gap), reconnect with EventsAfter using
	// afterWorkerSequence = ReplayGap.GetOldestAvailableSequence() - 1.
	//
	// The gateway sets ReplayGap only on StreamEvents results, never on a normal
	// (non-resumed) stream and never on DrainEvents.
	ReplayGap *ReplayGap
	// Err is the terminal stream error; when non-nil no further results follow.
	Err error
}

// IsReplayGap reports whether this result carries the gateway's reconnect-replay
// gap sentinel rather than a normal event or a terminal error.
func (r EventResult) IsReplayGap() bool {
	return r.ReplayGap != nil
}

// EventSubscription owns a running gateway event stream.
type EventSubscription struct {
	results <-chan EventResult
	cancel  context.CancelFunc
	done    <-chan struct{}
	once    sync.Once
}

// Events returns the stream results channel.
func (s *EventSubscription) Events() <-chan EventResult {
	return s.results
}

// Close cancels the stream and waits for the receive goroutine to stop.
func (s *EventSubscription) Close() {
	if s == nil {
		return
	}
	s.once.Do(func() {
		s.cancel()
		<-s.done
	})
}

// Session represents one gateway-backed MXAccess session.
type Session struct {
	client     *Client
	openReply  *OpenSessionReply
	closeMu    sync.Mutex
	closeReply *CloseSessionReply
}

func newSession(client *Client, openReply *OpenSessionReply) *Session {
	return &Session{
		client:    client,
		openReply: openReply,
	}
}

// NewSessionForID creates a session wrapper for commands against an existing
// gateway session id.
func NewSessionForID(client *Client, sessionID string) *Session {
	return newSession(client, &pb.OpenSessionReply{SessionId: sessionID})
}

// ID returns the gateway session identifier.
func (s *Session) ID() string {
	return s.openReply.GetSessionId()
}

// OpenReply returns the raw OpenSession reply.
func (s *Session) OpenReply() *OpenSessionReply {
	return s.openReply
}

// Close closes the gateway session once and returns the raw close reply.
func (s *Session) Close(ctx context.Context) (*CloseSessionReply, error) {
	s.closeMu.Lock()
	defer s.closeMu.Unlock()

	if s.closeReply != nil {
		return s.closeReply, nil
	}

	reply, err := s.client.CloseSessionRaw(ctx, &pb.CloseSessionRequest{SessionId: s.ID()})
	if err != nil {
		return reply, err
	}
	s.closeReply = reply
	return reply, nil
}

// Register invokes MXAccess Register and returns the server handle.
func (s *Session) Register(ctx context.Context, clientName string) (int32, error) {
	reply, err := s.RegisterRaw(ctx, clientName)
	if err != nil {
		return 0, err
	}
	if reply.GetRegister() != nil {
		return reply.GetRegister().GetServerHandle(), nil
	}
	return reply.GetReturnValue().GetInt32Value(), nil
}

// RegisterRaw invokes MXAccess Register and returns the raw reply.
func (s *Session) RegisterRaw(ctx context.Context, clientName string) (*MxCommandReply, error) {
	if clientName == "" {
		return nil, errors.New("mxgateway: client name is required")
	}

	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_REGISTER,
		Payload: &pb.MxCommand_Register{
			Register: &pb.RegisterCommand{ClientName: clientName},
		},
	})
}

// Unregister invokes MXAccess Unregister.
func (s *Session) Unregister(ctx context.Context, serverHandle int32) error {
	_, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_UNREGISTER,
		Payload: &pb.MxCommand_Unregister{
			Unregister: &pb.UnregisterCommand{ServerHandle: serverHandle},
		},
	})
	return err
}

// RemoveItem invokes MXAccess RemoveItem.
func (s *Session) RemoveItem(ctx context.Context, serverHandle, itemHandle int32) error {
	_, err := s.RemoveItemRaw(ctx, serverHandle, itemHandle)
	return err
}

// RemoveItemRaw invokes MXAccess RemoveItem and returns the raw reply.
func (s *Session) RemoveItemRaw(ctx context.Context, serverHandle, itemHandle int32) (*MxCommandReply, error) {
	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_REMOVE_ITEM,
		Payload: &pb.MxCommand_RemoveItem{
			RemoveItem: &pb.RemoveItemCommand{
				ServerHandle: serverHandle,
				ItemHandle:   itemHandle,
			},
		},
	})
}

// AddItem invokes MXAccess AddItem and returns the item handle.
func (s *Session) AddItem(ctx context.Context, serverHandle int32, itemDefinition string) (int32, error) {
	reply, err := s.AddItemRaw(ctx, serverHandle, itemDefinition)
	if err != nil {
		return 0, err
	}
	if reply.GetAddItem() != nil {
		return reply.GetAddItem().GetItemHandle(), nil
	}
	return reply.GetReturnValue().GetInt32Value(), nil
}

// AddItemRaw invokes MXAccess AddItem and returns the raw reply.
func (s *Session) AddItemRaw(ctx context.Context, serverHandle int32, itemDefinition string) (*MxCommandReply, error) {
	if itemDefinition == "" {
		return nil, errors.New("mxgateway: item definition is required")
	}

	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_ADD_ITEM,
		Payload: &pb.MxCommand_AddItem{
			AddItem: &pb.AddItemCommand{
				ServerHandle:   serverHandle,
				ItemDefinition: itemDefinition,
			},
		},
	})
}

// AddItem2 invokes MXAccess AddItem2 and returns the item handle.
func (s *Session) AddItem2(ctx context.Context, serverHandle int32, itemDefinition, itemContext string) (int32, error) {
	reply, err := s.AddItem2Raw(ctx, serverHandle, itemDefinition, itemContext)
	if err != nil {
		return 0, err
	}
	if reply.GetAddItem2() != nil {
		return reply.GetAddItem2().GetItemHandle(), nil
	}
	return reply.GetReturnValue().GetInt32Value(), nil
}

// AddItem2Raw invokes MXAccess AddItem2 and returns the raw reply.
func (s *Session) AddItem2Raw(ctx context.Context, serverHandle int32, itemDefinition, itemContext string) (*MxCommandReply, error) {
	if itemDefinition == "" {
		return nil, errors.New("mxgateway: item definition is required")
	}

	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_ADD_ITEM2,
		Payload: &pb.MxCommand_AddItem2{
			AddItem2: &pb.AddItem2Command{
				ServerHandle:   serverHandle,
				ItemDefinition: itemDefinition,
				ItemContext:    itemContext,
			},
		},
	})
}

// Advise invokes MXAccess Advise.
func (s *Session) Advise(ctx context.Context, serverHandle, itemHandle int32) error {
	_, err := s.AdviseRaw(ctx, serverHandle, itemHandle)
	return err
}

// AdviseRaw invokes MXAccess Advise and returns the raw reply.
func (s *Session) AdviseRaw(ctx context.Context, serverHandle, itemHandle int32) (*MxCommandReply, error) {
	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_ADVISE,
		Payload: &pb.MxCommand_Advise{
			Advise: &pb.AdviseCommand{
				ServerHandle: serverHandle,
				ItemHandle:   itemHandle,
			},
		},
	})
}

// UnAdvise invokes MXAccess UnAdvise.
func (s *Session) UnAdvise(ctx context.Context, serverHandle, itemHandle int32) error {
	_, err := s.UnAdviseRaw(ctx, serverHandle, itemHandle)
	return err
}

// UnAdviseRaw invokes MXAccess UnAdvise and returns the raw reply.
func (s *Session) UnAdviseRaw(ctx context.Context, serverHandle, itemHandle int32) (*MxCommandReply, error) {
	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_UN_ADVISE,
		Payload: &pb.MxCommand_UnAdvise{
			UnAdvise: &pb.UnAdviseCommand{
				ServerHandle: serverHandle,
				ItemHandle:   itemHandle,
			},
		},
	})
}

// AddItemBulk invokes MXAccess AddItem for each tag inside one gateway command.
func (s *Session) AddItemBulk(ctx context.Context, serverHandle int32, tagAddresses []string) ([]*SubscribeResult, error) {
	if tagAddresses == nil {
		return nil, errors.New("mxgateway: tag addresses are required")
	}
	if err := ensureBulkSize("tag addresses", len(tagAddresses)); err != nil {
		return nil, err
	}
	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_ADD_ITEM_BULK,
		Payload: &pb.MxCommand_AddItemBulk{
			AddItemBulk: &pb.AddItemBulkCommand{
				ServerHandle: serverHandle,
				TagAddresses: tagAddresses,
			},
		},
	})
	if err != nil {
		return nil, err
	}
	return reply.GetAddItemBulk().GetResults(), nil
}

// AdviseItemBulk invokes MXAccess Advise for each item handle inside one gateway command.
func (s *Session) AdviseItemBulk(ctx context.Context, serverHandle int32, itemHandles []int32) ([]*SubscribeResult, error) {
	if itemHandles == nil {
		return nil, errors.New("mxgateway: item handles are required")
	}
	if err := ensureBulkSize("item handles", len(itemHandles)); err != nil {
		return nil, err
	}
	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_ADVISE_ITEM_BULK,
		Payload: &pb.MxCommand_AdviseItemBulk{
			AdviseItemBulk: &pb.AdviseItemBulkCommand{
				ServerHandle: serverHandle,
				ItemHandles:  itemHandles,
			},
		},
	})
	if err != nil {
		return nil, err
	}
	return reply.GetAdviseItemBulk().GetResults(), nil
}

// RemoveItemBulk invokes MXAccess RemoveItem for each item handle inside one gateway command.
func (s *Session) RemoveItemBulk(ctx context.Context, serverHandle int32, itemHandles []int32) ([]*SubscribeResult, error) {
	if itemHandles == nil {
		return nil, errors.New("mxgateway: item handles are required")
	}
	if err := ensureBulkSize("item handles", len(itemHandles)); err != nil {
		return nil, err
	}
	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_REMOVE_ITEM_BULK,
		Payload: &pb.MxCommand_RemoveItemBulk{
			RemoveItemBulk: &pb.RemoveItemBulkCommand{
				ServerHandle: serverHandle,
				ItemHandles:  itemHandles,
			},
		},
	})
	if err != nil {
		return nil, err
	}
	return reply.GetRemoveItemBulk().GetResults(), nil
}

// UnAdviseItemBulk invokes MXAccess UnAdvise for each item handle inside one gateway command.
func (s *Session) UnAdviseItemBulk(ctx context.Context, serverHandle int32, itemHandles []int32) ([]*SubscribeResult, error) {
	if itemHandles == nil {
		return nil, errors.New("mxgateway: item handles are required")
	}
	if err := ensureBulkSize("item handles", len(itemHandles)); err != nil {
		return nil, err
	}
	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_UN_ADVISE_ITEM_BULK,
		Payload: &pb.MxCommand_UnAdviseItemBulk{
			UnAdviseItemBulk: &pb.UnAdviseItemBulkCommand{
				ServerHandle: serverHandle,
				ItemHandles:  itemHandles,
			},
		},
	})
	if err != nil {
		return nil, err
	}
	return reply.GetUnAdviseItemBulk().GetResults(), nil
}

// SubscribeBulk invokes AddItem and Advise for each tag inside one gateway command.
func (s *Session) SubscribeBulk(ctx context.Context, serverHandle int32, tagAddresses []string) ([]*SubscribeResult, error) {
	if tagAddresses == nil {
		return nil, errors.New("mxgateway: tag addresses are required")
	}
	if err := ensureBulkSize("tag addresses", len(tagAddresses)); err != nil {
		return nil, err
	}
	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_SUBSCRIBE_BULK,
		Payload: &pb.MxCommand_SubscribeBulk{
			SubscribeBulk: &pb.SubscribeBulkCommand{
				ServerHandle: serverHandle,
				TagAddresses: tagAddresses,
			},
		},
	})
	if err != nil {
		return nil, err
	}
	return reply.GetSubscribeBulk().GetResults(), nil
}

// UnsubscribeBulk invokes UnAdvise and RemoveItem for each item handle inside one gateway command.
func (s *Session) UnsubscribeBulk(ctx context.Context, serverHandle int32, itemHandles []int32) ([]*SubscribeResult, error) {
	if itemHandles == nil {
		return nil, errors.New("mxgateway: item handles are required")
	}
	if err := ensureBulkSize("item handles", len(itemHandles)); err != nil {
		return nil, err
	}
	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_UNSUBSCRIBE_BULK,
		Payload: &pb.MxCommand_UnsubscribeBulk{
			UnsubscribeBulk: &pb.UnsubscribeBulkCommand{
				ServerHandle: serverHandle,
				ItemHandles:  itemHandles,
			},
		},
	})
	if err != nil {
		return nil, err
	}
	return reply.GetUnsubscribeBulk().GetResults(), nil
}

// WriteBulk invokes MXAccess Write sequentially for each entry inside one gateway command.
// Per-entry failures appear as BulkWriteResult entries with WasSuccessful=false; the call
// never returns an error for per-entry MXAccess failures (it returns an error only for
// protocol-level failures or transport errors).
//
// A non-nil but empty entries slice is treated as a no-op and returns an empty result
// without a wire round-trip; pass nil to surface a clear "entries are required" error.
func (s *Session) WriteBulk(ctx context.Context, serverHandle int32, entries []*WriteBulkEntry) ([]*BulkWriteResult, error) {
	if entries == nil {
		return nil, errors.New("mxgateway: write bulk entries are required")
	}
	if err := ensureBulkSize("write bulk entries", len(entries)); err != nil {
		return nil, err
	}
	if len(entries) == 0 {
		return []*BulkWriteResult{}, nil
	}
	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_WRITE_BULK,
		Payload: &pb.MxCommand_WriteBulk{
			WriteBulk: &pb.WriteBulkCommand{
				ServerHandle: serverHandle,
				Entries:      entries,
			},
		},
	})
	if err != nil {
		return nil, err
	}
	return reply.GetWriteBulk().GetResults(), nil
}

// Write2Bulk invokes MXAccess Write2 (timestamped) for each entry inside one gateway command.
//
// A non-nil but empty entries slice is treated as a no-op and returns an empty result
// without a wire round-trip; pass nil to surface a clear "entries are required" error.
func (s *Session) Write2Bulk(ctx context.Context, serverHandle int32, entries []*Write2BulkEntry) ([]*BulkWriteResult, error) {
	if entries == nil {
		return nil, errors.New("mxgateway: write2 bulk entries are required")
	}
	if err := ensureBulkSize("write2 bulk entries", len(entries)); err != nil {
		return nil, err
	}
	if len(entries) == 0 {
		return []*BulkWriteResult{}, nil
	}
	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_WRITE2_BULK,
		Payload: &pb.MxCommand_Write2Bulk{
			Write2Bulk: &pb.Write2BulkCommand{
				ServerHandle: serverHandle,
				Entries:      entries,
			},
		},
	})
	if err != nil {
		return nil, err
	}
	return reply.GetWrite2Bulk().GetResults(), nil
}

// WriteSecuredBulk invokes MXAccess WriteSecured for each entry. Credential-sensitive
// values must not be logged by callers; mirrors the single-item WriteSecured contract.
//
// A non-nil but empty entries slice is treated as a no-op and returns an empty result
// without a wire round-trip; pass nil to surface a clear "entries are required" error.
func (s *Session) WriteSecuredBulk(ctx context.Context, serverHandle int32, entries []*WriteSecuredBulkEntry) ([]*BulkWriteResult, error) {
	if entries == nil {
		return nil, errors.New("mxgateway: write-secured bulk entries are required")
	}
	if err := ensureBulkSize("write-secured bulk entries", len(entries)); err != nil {
		return nil, err
	}
	if len(entries) == 0 {
		return []*BulkWriteResult{}, nil
	}
	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_WRITE_SECURED_BULK,
		Payload: &pb.MxCommand_WriteSecuredBulk{
			WriteSecuredBulk: &pb.WriteSecuredBulkCommand{
				ServerHandle: serverHandle,
				Entries:      entries,
			},
		},
	})
	if err != nil {
		return nil, err
	}
	return reply.GetWriteSecuredBulk().GetResults(), nil
}

// WriteSecured2Bulk invokes MXAccess WriteSecured2 (timestamped) for each entry.
//
// A non-nil but empty entries slice is treated as a no-op and returns an empty result
// without a wire round-trip; pass nil to surface a clear "entries are required" error.
func (s *Session) WriteSecured2Bulk(ctx context.Context, serverHandle int32, entries []*WriteSecured2BulkEntry) ([]*BulkWriteResult, error) {
	if entries == nil {
		return nil, errors.New("mxgateway: write-secured2 bulk entries are required")
	}
	if err := ensureBulkSize("write-secured2 bulk entries", len(entries)); err != nil {
		return nil, err
	}
	if len(entries) == 0 {
		return []*BulkWriteResult{}, nil
	}
	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_WRITE_SECURED2_BULK,
		Payload: &pb.MxCommand_WriteSecured2Bulk{
			WriteSecured2Bulk: &pb.WriteSecured2BulkCommand{
				ServerHandle: serverHandle,
				Entries:      entries,
			},
		},
	})
	if err != nil {
		return nil, err
	}
	return reply.GetWriteSecured2Bulk().GetResults(), nil
}

// ReadBulk snapshots the current value of each requested tag.
//
// MXAccess COM has no synchronous Read; the worker satisfies this by returning the
// most recent cached OnDataChange value when the tag is already advised (WasCached=true),
// or by taking a full AddItem + Advise + wait + UnAdvise + RemoveItem snapshot lifecycle
// otherwise. timeout bounds the wait per tag in the snapshot case; pass zero to use the
// worker default. Per-tag failures (timeout, invalid tag) appear as BulkReadResult entries
// with WasSuccessful=false; the call never returns an error for per-tag MXAccess failures.
//
// A non-nil but empty tagAddresses slice is treated as a no-op and returns an empty
// result without a wire round-trip; pass nil to surface a clear "tag addresses are
// required" error.
func (s *Session) ReadBulk(ctx context.Context, serverHandle int32, tagAddresses []string, timeout time.Duration) ([]*BulkReadResult, error) {
	if tagAddresses == nil {
		return nil, errors.New("mxgateway: tag addresses are required")
	}
	if err := ensureBulkSize("tag addresses", len(tagAddresses)); err != nil {
		return nil, err
	}
	if len(tagAddresses) == 0 {
		return []*BulkReadResult{}, nil
	}
	var timeoutMs uint32
	if timeout > 0 {
		ms := timeout.Milliseconds()
		if ms > int64(^uint32(0)) {
			timeoutMs = ^uint32(0)
		} else {
			timeoutMs = uint32(ms)
		}
	}
	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_READ_BULK,
		Payload: &pb.MxCommand_ReadBulk{
			ReadBulk: &pb.ReadBulkCommand{
				ServerHandle: serverHandle,
				TagAddresses: tagAddresses,
				TimeoutMs:    timeoutMs,
			},
		},
	})
	if err != nil {
		return nil, err
	}
	return reply.GetReadBulk().GetResults(), nil
}

// Write invokes MXAccess Write.
func (s *Session) Write(ctx context.Context, serverHandle, itemHandle int32, value *MxValue, userID int32) error {
	_, err := s.WriteRaw(ctx, serverHandle, itemHandle, value, userID)
	return err
}

// WriteRaw invokes MXAccess Write and returns the raw reply.
func (s *Session) WriteRaw(ctx context.Context, serverHandle, itemHandle int32, value *MxValue, userID int32) (*MxCommandReply, error) {
	if value == nil {
		return nil, errors.New("mxgateway: write value is required")
	}

	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_WRITE,
		Payload: &pb.MxCommand_Write{
			Write: &pb.WriteCommand{
				ServerHandle: serverHandle,
				ItemHandle:   itemHandle,
				Value:        value,
				UserId:       userID,
			},
		},
	})
}

// WriteArrayElements writes a sparse, default-filled array: only the given
// elements (index → scalar value) are set; every unmentioned index up to
// totalLength is written as the element type's default (false / 0 / "" / Unix
// epoch for time). The gateway expands the sparse representation into a full
// array write before forwarding to MXAccess — this is a RESET of unmentioned
// indices, not a preserve. Neither RESET semantics nor the original array
// content are retained.
//
// elementDataType must be a scalar MXAccess type (Boolean, Integer, Float,
// Double, String, or Time). totalLength must be at least as large as the
// highest index in elements plus one.
func (s *Session) WriteArrayElements(
	ctx context.Context,
	serverHandle, itemHandle int32,
	elementDataType MxDataType,
	totalLength uint32,
	elements map[uint32]*MxValue,
	userID int32,
) error {
	return s.Write(ctx, serverHandle, itemHandle, buildSparseArrayValue(elementDataType, totalLength, elements), userID)
}

// buildSparseArrayValue constructs the MxValue carrying an MxSparseArray oneof
// arm from a map of index → scalar MxValue. Keys are visited in ascending
// order so the produced slice is deterministic (important for test assertions).
func buildSparseArrayValue(elementDataType MxDataType, totalLength uint32, elements map[uint32]*MxValue) *MxValue {
	indices := make([]uint32, 0, len(elements))
	for idx := range elements {
		indices = append(indices, idx)
	}
	sort.Slice(indices, func(i, j int) bool { return indices[i] < indices[j] })

	sparseElements := make([]*MxSparseElement, 0, len(elements))
	for _, idx := range indices {
		sparseElements = append(sparseElements, &MxSparseElement{
			Index: idx,
			Value: elements[idx],
		})
	}

	return &MxValue{
		Kind: &pb.MxValue_SparseArrayValue{
			SparseArrayValue: &MxSparseArray{
				ElementDataType: elementDataType,
				TotalLength:     totalLength,
				Elements:        sparseElements,
			},
		},
	}
}

// PingRaw sends a diagnostic PING command and returns the raw reply.
// The message is echoed back by the gateway in the reply's DiagnosticMessage field.
func (s *Session) PingRaw(ctx context.Context, message string) (*MxCommandReply, error) {
	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_PING,
		Payload: &pb.MxCommand_Ping{
			Ping: &pb.PingCommand{Message: message},
		},
	})
}

// Write2 invokes MXAccess Write2 (timestamped single-item write).
func (s *Session) Write2(ctx context.Context, serverHandle, itemHandle int32, value, timestampValue *MxValue, userID int32) error {
	_, err := s.Write2Raw(ctx, serverHandle, itemHandle, value, timestampValue, userID)
	return err
}

// Write2Raw invokes MXAccess Write2 (timestamped single-item write) and returns the raw reply.
func (s *Session) Write2Raw(ctx context.Context, serverHandle, itemHandle int32, value, timestampValue *MxValue, userID int32) (*MxCommandReply, error) {
	if value == nil {
		return nil, errors.New("mxgateway: write2 value is required")
	}
	if timestampValue == nil {
		return nil, errors.New("mxgateway: write2 timestamp value is required")
	}

	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_WRITE2,
		Payload: &pb.MxCommand_Write2{
			Write2: &pb.Write2Command{
				ServerHandle:   serverHandle,
				ItemHandle:     itemHandle,
				Value:          value,
				TimestampValue: timestampValue,
				UserId:         userID,
			},
		},
	})
}

// AdviseSupervisory invokes MXAccess AdviseSupervisory, advising an item on the
// supervisory (as opposed to runtime) data path.
func (s *Session) AdviseSupervisory(ctx context.Context, serverHandle, itemHandle int32) error {
	_, err := s.AdviseSupervisoryRaw(ctx, serverHandle, itemHandle)
	return err
}

// AdviseSupervisoryRaw invokes MXAccess AdviseSupervisory and returns the raw reply.
func (s *Session) AdviseSupervisoryRaw(ctx context.Context, serverHandle, itemHandle int32) (*MxCommandReply, error) {
	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_ADVISE_SUPERVISORY,
		Payload: &pb.MxCommand_AdviseSupervisory{
			AdviseSupervisory: &pb.AdviseSupervisoryCommand{
				ServerHandle: serverHandle,
				ItemHandle:   itemHandle,
			},
		},
	})
}

// WriteSecured invokes MXAccess WriteSecured (secured single-item write).
//
// The value is credential-sensitive: callers must not log it, and any error this
// call surfaces has the value's string content redacted (see WriteSecuredRaw).
// MXAccess parity is preserved — WriteSecured legitimately fails when it is not
// preceded by a matching AuthenticateUser + AdviseSupervisory or when the body
// carries no value; that native failure is surfaced as-is, never pre-empted or
// reordered by this client.
func (s *Session) WriteSecured(ctx context.Context, serverHandle, itemHandle, currentUserID, verifierUserID int32, value *MxValue) error {
	_, err := s.WriteSecuredRaw(ctx, serverHandle, itemHandle, currentUserID, verifierUserID, value)
	return err
}

// WriteSecuredRaw invokes MXAccess WriteSecured and returns the raw reply. Any
// surfaced error is routed through the client's secret-redaction seam so a string
// write value never appears in error text.
func (s *Session) WriteSecuredRaw(ctx context.Context, serverHandle, itemHandle, currentUserID, verifierUserID int32, value *MxValue) (*MxCommandReply, error) {
	if value == nil {
		return nil, errors.New("mxgateway: write-secured value is required")
	}

	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_WRITE_SECURED,
		Payload: &pb.MxCommand_WriteSecured{
			WriteSecured: &pb.WriteSecuredCommand{
				ServerHandle:   serverHandle,
				ItemHandle:     itemHandle,
				CurrentUserId:  currentUserID,
				VerifierUserId: verifierUserID,
				Value:          value,
			},
		},
	})
	return reply, redactSecrets(err, stringSecrets(value)...)
}

// WriteSecured2 invokes MXAccess WriteSecured2 (secured, timestamped single-item write).
//
// Like WriteSecured, the value is credential-sensitive and its string content is
// scrubbed from any surfaced error. Native parity failures (missing prior
// AuthenticateUser/AdviseSupervisory, value-less body) are surfaced unchanged.
func (s *Session) WriteSecured2(ctx context.Context, serverHandle, itemHandle, currentUserID, verifierUserID int32, value, timestampValue *MxValue) error {
	_, err := s.WriteSecured2Raw(ctx, serverHandle, itemHandle, currentUserID, verifierUserID, value, timestampValue)
	return err
}

// WriteSecured2Raw invokes MXAccess WriteSecured2 and returns the raw reply. Any
// surfaced error is routed through the client's secret-redaction seam.
func (s *Session) WriteSecured2Raw(ctx context.Context, serverHandle, itemHandle, currentUserID, verifierUserID int32, value, timestampValue *MxValue) (*MxCommandReply, error) {
	if value == nil {
		return nil, errors.New("mxgateway: write-secured2 value is required")
	}
	if timestampValue == nil {
		return nil, errors.New("mxgateway: write-secured2 timestamp value is required")
	}

	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_WRITE_SECURED2,
		Payload: &pb.MxCommand_WriteSecured2{
			WriteSecured2: &pb.WriteSecured2Command{
				ServerHandle:   serverHandle,
				ItemHandle:     itemHandle,
				CurrentUserId:  currentUserID,
				VerifierUserId: verifierUserID,
				Value:          value,
				TimestampValue: timestampValue,
			},
		},
	})
	return reply, redactSecrets(err, stringSecrets(value)...)
}

// AuthenticateUser invokes MXAccess AuthenticateUser and returns the resolved
// MXAccess user id.
//
// verifyUserPassword is a raw MXAccess credential: this client never logs it and
// scrubs it from any error it surfaces (see AuthenticateUserRaw). Callers must
// likewise keep it out of their own logs, metrics, and diagnostics.
func (s *Session) AuthenticateUser(ctx context.Context, serverHandle int32, verifyUser, verifyUserPassword string) (int32, error) {
	reply, err := s.AuthenticateUserRaw(ctx, serverHandle, verifyUser, verifyUserPassword)
	if err != nil {
		return 0, err
	}
	if reply.GetAuthenticateUser() != nil {
		return reply.GetAuthenticateUser().GetUserId(), nil
	}
	return reply.GetReturnValue().GetInt32Value(), nil
}

// AuthenticateUserRaw invokes MXAccess AuthenticateUser and returns the raw
// reply. The credential is scrubbed from any surfaced error via the client's
// secret-redaction seam, so even a gateway diagnostic echoing the password back
// cannot leak it through this call's error.
func (s *Session) AuthenticateUserRaw(ctx context.Context, serverHandle int32, verifyUser, verifyUserPassword string) (*MxCommandReply, error) {
	if verifyUser == "" {
		return nil, errors.New("mxgateway: verify user is required")
	}

	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_AUTHENTICATE_USER,
		Payload: &pb.MxCommand_AuthenticateUser{
			AuthenticateUser: &pb.AuthenticateUserCommand{
				ServerHandle:       serverHandle,
				VerifyUser:         verifyUser,
				VerifyUserPassword: verifyUserPassword,
			},
		},
	})
	return reply, redactSecrets(err, verifyUserPassword)
}

// ArchestrAUserToId invokes MXAccess ArchestrAUserToId, resolving an ArchestrA
// user GUID to its MXAccess integer user id.
func (s *Session) ArchestrAUserToId(ctx context.Context, serverHandle int32, userIDGuid string) (int32, error) {
	reply, err := s.ArchestrAUserToIdRaw(ctx, serverHandle, userIDGuid)
	if err != nil {
		return 0, err
	}
	if reply.GetArchestraUserToId() != nil {
		return reply.GetArchestraUserToId().GetUserId(), nil
	}
	return reply.GetReturnValue().GetInt32Value(), nil
}

// ArchestrAUserToIdRaw invokes MXAccess ArchestrAUserToId and returns the raw reply.
func (s *Session) ArchestrAUserToIdRaw(ctx context.Context, serverHandle int32, userIDGuid string) (*MxCommandReply, error) {
	if userIDGuid == "" {
		return nil, errors.New("mxgateway: user id GUID is required")
	}

	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_ARCHESTRA_USER_TO_ID,
		Payload: &pb.MxCommand_ArchestraUserToId{
			ArchestraUserToId: &pb.ArchestrAUserToIdCommand{
				ServerHandle: serverHandle,
				UserIdGuid:   userIDGuid,
			},
		},
	})
}

// AddBufferedItem invokes MXAccess AddBufferedItem and returns the item handle.
func (s *Session) AddBufferedItem(ctx context.Context, serverHandle int32, itemDefinition, itemContext string) (int32, error) {
	reply, err := s.AddBufferedItemRaw(ctx, serverHandle, itemDefinition, itemContext)
	if err != nil {
		return 0, err
	}
	if reply.GetAddBufferedItem() != nil {
		return reply.GetAddBufferedItem().GetItemHandle(), nil
	}
	return reply.GetReturnValue().GetInt32Value(), nil
}

// AddBufferedItemRaw invokes MXAccess AddBufferedItem and returns the raw reply.
func (s *Session) AddBufferedItemRaw(ctx context.Context, serverHandle int32, itemDefinition, itemContext string) (*MxCommandReply, error) {
	if itemDefinition == "" {
		return nil, errors.New("mxgateway: item definition is required")
	}

	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_ADD_BUFFERED_ITEM,
		Payload: &pb.MxCommand_AddBufferedItem{
			AddBufferedItem: &pb.AddBufferedItemCommand{
				ServerHandle:   serverHandle,
				ItemDefinition: itemDefinition,
				ItemContext:    itemContext,
			},
		},
	})
}

// SetBufferedUpdateInterval invokes MXAccess SetBufferedUpdateInterval.
func (s *Session) SetBufferedUpdateInterval(ctx context.Context, serverHandle, updateIntervalMilliseconds int32) error {
	_, err := s.SetBufferedUpdateIntervalRaw(ctx, serverHandle, updateIntervalMilliseconds)
	return err
}

// SetBufferedUpdateIntervalRaw invokes MXAccess SetBufferedUpdateInterval and returns the raw reply.
func (s *Session) SetBufferedUpdateIntervalRaw(ctx context.Context, serverHandle, updateIntervalMilliseconds int32) (*MxCommandReply, error) {
	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_SET_BUFFERED_UPDATE_INTERVAL,
		Payload: &pb.MxCommand_SetBufferedUpdateInterval{
			SetBufferedUpdateInterval: &pb.SetBufferedUpdateIntervalCommand{
				ServerHandle:               serverHandle,
				UpdateIntervalMilliseconds: updateIntervalMilliseconds,
			},
		},
	})
}

// Suspend invokes MXAccess Suspend and returns the resulting item status.
func (s *Session) Suspend(ctx context.Context, serverHandle, itemHandle int32) (*MxStatusProxy, error) {
	reply, err := s.SuspendRaw(ctx, serverHandle, itemHandle)
	if err != nil {
		return nil, err
	}
	return reply.GetSuspend().GetStatus(), nil
}

// SuspendRaw invokes MXAccess Suspend and returns the raw reply.
func (s *Session) SuspendRaw(ctx context.Context, serverHandle, itemHandle int32) (*MxCommandReply, error) {
	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_SUSPEND,
		Payload: &pb.MxCommand_Suspend{
			Suspend: &pb.SuspendCommand{
				ServerHandle: serverHandle,
				ItemHandle:   itemHandle,
			},
		},
	})
}

// Activate invokes MXAccess Activate and returns the resulting item status.
func (s *Session) Activate(ctx context.Context, serverHandle, itemHandle int32) (*MxStatusProxy, error) {
	reply, err := s.ActivateRaw(ctx, serverHandle, itemHandle)
	if err != nil {
		return nil, err
	}
	return reply.GetActivate().GetStatus(), nil
}

// ActivateRaw invokes MXAccess Activate and returns the raw reply.
func (s *Session) ActivateRaw(ctx context.Context, serverHandle, itemHandle int32) (*MxCommandReply, error) {
	return s.invokeCommand(ctx, &pb.MxCommand{
		Kind: pb.MxCommandKind_MX_COMMAND_KIND_ACTIVATE,
		Payload: &pb.MxCommand_Activate{
			Activate: &pb.ActivateCommand{
				ServerHandle: serverHandle,
				ItemHandle:   itemHandle,
			},
		},
	})
}

// stringSecrets collects the non-empty string content of the given values so it
// can be scrubbed from surfaced errors. Only string-typed MxValues carry
// scrubable text; non-string values (numbers, timestamps, arrays) contribute
// nothing.
func stringSecrets(values ...*MxValue) []string {
	var secrets []string
	for _, value := range values {
		if value == nil {
			continue
		}
		if stringValue, ok := value.GetKind().(*pb.MxValue_StringValue); ok && stringValue.StringValue != "" {
			secrets = append(secrets, stringValue.StringValue)
		}
	}
	return secrets
}

// Events streams ordered session events until the server ends the stream,
// context cancellation stops Recv, or a terminal error is sent.
//
// The returned channel is buffered. If the consumer falls behind and the buffer
// overflows, the stream is terminated and a final EventResult carrying a
// GatewayError that wraps ErrSlowConsumer is delivered before the channel
// closes. Callers must match it with errors.Is(res.Err, ErrSlowConsumer) to
// distinguish a slow-consumer drop from a graceful server end. Use
// SubscribeEvents for a blocking, backpressured stream that never drops.
func (s *Session) Events(ctx context.Context) (<-chan EventResult, error) {
	return s.EventsAfter(ctx, 0)
}

// EventsAfter streams ordered session events after the given worker sequence.
//
// Like Events, the returned channel is buffered and terminates with a final
// EventResult wrapping ErrSlowConsumer (matchable via errors.Is) if the consumer
// falls behind and the buffer overflows, rather than silently closing.
func (s *Session) EventsAfter(ctx context.Context, afterWorkerSequence uint64) (<-chan EventResult, error) {
	subscription, err := s.subscribeEventsAfter(ctx, afterWorkerSequence, true)
	if err != nil {
		return nil, err
	}
	return subscription.Events(), nil
}

// SubscribeEvents starts an owned event subscription.
func (s *Session) SubscribeEvents(ctx context.Context) (*EventSubscription, error) {
	return s.SubscribeEventsAfter(ctx, 0)
}

// SubscribeEventsAfter starts an owned event subscription after the given worker sequence.
func (s *Session) SubscribeEventsAfter(ctx context.Context, afterWorkerSequence uint64) (*EventSubscription, error) {
	return s.subscribeEventsAfter(ctx, afterWorkerSequence, false)
}

func (s *Session) subscribeEventsAfter(ctx context.Context, afterWorkerSequence uint64, cancelWhenResultBufferFull bool) (*EventSubscription, error) {
	streamCtx, cancel := context.WithCancel(ctx)
	stream, err := s.client.StreamEventsRaw(streamCtx, &pb.StreamEventsRequest{
		SessionId:           s.ID(),
		AfterWorkerSequence: afterWorkerSequence,
	})
	if err != nil {
		cancel()
		return nil, err
	}

	results := make(chan EventResult, eventBufferSize+eventBufferReservedSlots)
	done := make(chan struct{})
	go func() {
		defer close(results)
		defer close(done)
		for {
			event, err := stream.Recv()
			if err == nil {
				result := EventResult{Event: event}
				// The gateway marks a reconnect-replay gap with a sentinel MxEvent
				// carrying replay_gap (family UNSPECIFIED, body unset). Surface it
				// as a distinct typed signal rather than a normal event: clear
				// Event so consumers never process the sentinel as a data change.
				if gap := event.GetReplayGap(); gap != nil {
					result = EventResult{ReplayGap: gap}
				}
				if !sendEventResult(streamCtx, results, result, cancelWhenResultBufferFull, cancel) {
					return
				}
				continue
			}
			if err == io.EOF || status.Code(err) == codes.Canceled || streamCtx.Err() != nil {
				return
			}
			sendEventResult(
				streamCtx,
				results,
				EventResult{Err: &GatewayError{Op: "stream events", Err: err}},
				cancelWhenResultBufferFull,
				cancel)
			return
		}
	}()

	return &EventSubscription{
		results: results,
		cancel:  cancel,
		done:    done,
	}, nil
}

func ensureBulkSize(name string, length int) error {
	if length > maxBulkItems {
		return fmt.Errorf("mxgateway: %s bulk commands are limited to %d item(s)", name, maxBulkItems)
	}
	return nil
}

func sendEventResult(
	ctx context.Context,
	results chan<- EventResult,
	result EventResult,
	cancelWhenBufferFull bool,
	cancel context.CancelFunc,
) bool {
	if cancelWhenBufferFull {
		// Treat the channel as full once the eventBufferSize data slots are
		// occupied, keeping eventBufferReservedSlots free for the terminal
		// error. This goroutine is the sole producer, so len(results) only
		// grows by our own sends and shrinks as the consumer reads; the reserve
		// therefore always survives to carry ErrSlowConsumer. A plain buffered
		// send would instead consume the reserved slot as ordinary data.
		if len(results) >= eventBufferSize {
			// The consumer fell behind and the data slots are full. Cancel the
			// stream, then use the reserved terminal slot to enqueue a loud
			// ErrSlowConsumer result so overflow is always observable rather
			// than a silently closed channel indistinguishable from a graceful
			// server end.
			cancel()
			select {
			case results <- EventResult{Err: &GatewayError{Op: "stream events", Err: ErrSlowConsumer}}:
			default:
			}
			return false
		}
		select {
		case results <- result:
			return true
		case <-ctx.Done():
			return false
		}
	}

	select {
	case results <- result:
		return true
	case <-ctx.Done():
		return false
	}
}

func (s *Session) invokeCommand(ctx context.Context, command *MxCommand) (*MxCommandReply, error) {
	return s.client.Invoke(ctx, &pb.MxCommandRequest{
		SessionId:           s.ID(),
		ClientCorrelationId: newCorrelationID(),
		Command:             command,
	})
}

func newCorrelationID() string {
	var buffer [16]byte
	if _, err := rand.Read(buffer[:]); err != nil {
		return ""
	}
	return hex.EncodeToString(buffer[:])
}