dingtalk-workspace-cli/internal/event/transport/protocol.go

// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package transport

import (
	"encoding/json"
	"fmt"

	dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
)

// FrameType discriminates the JSON frames flowing over the IPC channel.
// Wire values are stable strings — bumping any of these is a protocol
// breaking change requiring a coordinated bus + consume rollout.
type FrameType string

const (
	FrameTypeHello               FrameType = "hello"                 // consume → bus
	FrameTypeHelloAck            FrameType = "hello_ack"             // bus → consume
	FrameTypeEvent               FrameType = "event"                 // bus → consume
	FrameTypeHeartbeat           FrameType = "heartbeat"             // bidirectional
	FrameTypeSourceState         FrameType = "source_state"          // bus → consume
	FrameTypeBye                 FrameType = "bye"                   // bidirectional
	FrameTypeStatusReq           FrameType = "status_req"            // consume/ad-hoc → bus
	FrameTypeStatusResp          FrameType = "status_resp"           // bus → consume/ad-hoc
	FrameTypeConsumerStopReq     FrameType = "consumer_stop_req"     // ad-hoc → bus
	FrameTypeConsumerStopResp    FrameType = "consumer_stop_resp"    // bus → ad-hoc
	FrameTypeCredentialUpdate    FrameType = "credential_update"     // consume → bus
	FrameTypeCredentialUpdateAck FrameType = "credential_update_ack" // bus → consume
)

// CredentialMode declares that a consumer needs an additive credential
// negotiation before it can register for events. The zero value preserves the
// original protocol.
type CredentialMode string

const (
	CredentialModeRuntimeToken CredentialMode = "runtime_token"
	CapabilityRuntimeTokenV1                  = "runtime_token_v1"
)

// Hello is the first frame a consumer sends after dialing the bus. The bus
// uses event_types + filter to do server-side pushdown (plan §1 decision:
// only events matching this filter are written to this consumer's sendCh).
type Hello struct {
	Type        FrameType `json:"type"`
	ConsumerPID int       `json:"consumer_pid"`
	EventTypes  []string  `json:"event_types,omitempty"`  // wildcard list ("im.*", "approval.*"); empty = catch-all
	Filter      string    `json:"filter,omitempty"`       // optional regex over event_type
	SubscribeID string    `json:"subscribe_id,omitempty"` // optional personal subscription isolation key; empty = no subscribe_id filter
	Compact     bool      `json:"compact,omitempty"`      // hint to status output; bus does not transform payloads
	// Role distinguishes a real consumer (registered for events) from an
	// ad-hoc tooling connection (status/list/stop). Ad-hoc connections do
	// NOT register with the Hub.
	Role           HelloRole      `json:"role,omitempty"`
	CredentialMode CredentialMode `json:"credential_mode,omitempty"`
}

// HelloRole tags the purpose of a Hello connection.
type HelloRole string

const (
	HelloRoleConsumer     HelloRole = ""              // default
	HelloRoleStatus       HelloRole = "status"        // event list / event status
	HelloRoleStop         HelloRole = "stop"          // event stop (graceful trigger)
	HelloRoleConsumerStop HelloRole = "consumer_stop" // stop selected consumers only
)

// HelloAck is the bus's reply on accepted Hello. SourceState/StateSource
// mirror the source.Machine snapshot at connect time; CredentialsSource
// fields tell users which channel actually supplied the credentials in use
// (env / app_config / keychain / plain_config), see plan "凭证来源拆字段".
type HelloAck struct {
	Type               FrameType `json:"type"`
	BusPID             int       `json:"bus_pid"`
	SourceState        string    `json:"source_state"`                // mirrors source.State string value
	StateSource        string    `json:"state_source"`                // "hook" | "inferred"
	ClientIDSource     string    `json:"client_id_source"`            // auth.CredentialSource string
	ClientSecretSource string    `json:"client_secret_source"`        // auth.CredentialSource string
	IdleTimeoutSecs    int       `json:"idle_timeout_secs,omitempty"` // bus's IdleTimeout for diagnostics
	Capabilities       []string  `json:"capabilities,omitempty"`
	// CredentialGeneration is process-local and contains no secret data.
	CredentialGeneration uint64 `json:"credential_generation"`
	// TerminalReason is a fixed, non-sensitive bus terminal state. A runtime
	// client checks it before sending credential material.
	TerminalReason string `json:"terminal_reason,omitempty"`
}

// CredentialUpdate installs a host-supplied runtime token into a compatible
// bus after the bus has advertised CapabilityRuntimeTokenV1. Token is carried
// only over the owner-only local IPC connection and must never be logged.
type CredentialUpdate struct {
	Type               FrameType `json:"type"`
	ExpectedGeneration uint64    `json:"expected_generation"`
	Token              string    `json:"token"`
}

const (
	CredentialErrorGenerationConflict = "generation_conflict"
	CredentialErrorInvalid            = "invalid_credential"
	CredentialErrorRegistration       = "registration_failed"
	CredentialErrorRuntimeRejected    = "runtime_token_rejected"
	CredentialErrorInternal           = "internal_error"
)

// CredentialUpdateAck reports the CAS result without echoing any credential.
type CredentialUpdateAck struct {
	Type                 FrameType `json:"type"`
	Accepted             bool      `json:"accepted"`
	CredentialGeneration uint64    `json:"credential_generation"`
	ErrorCode            string    `json:"error_code,omitempty"`
	Error                string    `json:"error,omitempty"`
}

// Event wraps one delivered RawEvent for the wire. We keep the payload as
// a string (not nested JSON) so the bus does not need to parse / re-encode.
type Event struct {
	Type              FrameType         `json:"type"`
	Seq               uint64            `json:"seq"` // per-consumer monotonic, restarts at 1 on reconnect
	EventID           string            `json:"event_id"`
	EventBornTime     int64             `json:"event_born_time"`
	EventCorpID       string            `json:"event_corp_id,omitempty"`
	EventType         string            `json:"event_type"`
	EventUnifiedAppID string            `json:"event_unified_app_id,omitempty"`
	EventScope        string            `json:"event_scope,omitempty"`
	SubscribeID       string            `json:"subscribe_id,omitempty"`
	SourceID          string            `json:"source_id,omitempty"`
	RuleType          string            `json:"rule_type,omitempty"`
	Data              string            `json:"data"`
	Headers           map[string]string `json:"headers,omitempty"`
	ReceivedAtUnixMS  int64             `json:"received_at_unix_ms"`
}

// Heartbeat is bidirectional and stateless. It exists only to give both
// endpoints a chance to notice a dead peer via Read failure.
type Heartbeat struct {
	Type FrameType `json:"type"`
}

// SourceState is pushed bus → consume whenever the connection state machine
// transitions to / from connected. Consumers may render it; v1 they just
// forward to stderr when not --quiet.
type SourceState struct {
	Type        FrameType `json:"type"`
	State       string    `json:"state"`        // source.State string
	StateSource string    `json:"state_source"` // hook | inferred
	Attempt     int       `json:"attempt,omitempty"`
}

// Bye is sent by either side at graceful shutdown. Reason is free-form
// for logs; structured shutdown causes are encoded in the value:
//
//	"client_done"   — consume reached --max-events/--duration or SIGINT
//	"shutdown"      — bus SIGTERM/SIGINT
//	"idle_timeout"  — bus IdleTimeout fired with no consumers
//	"stop_request"  — bus received explicit Stop RPC
//	"subscription_stopped" — one consumer was removed by subscribe_id
type Bye struct {
	Type   FrameType `json:"type"`
	Reason string    `json:"reason"`
}

const (
	ByeReasonSubscriptionStopped  = "subscription_stopped"
	ByeReasonRuntimeTokenRejected = "runtime_token_rejected"
)

// ConsumerStopReq asks the bus to close consumers whose exact personal
// subscription IDs match. It is an additive local IPC control operation;
// the remote Stream connection remains alive while other consumers exist.
type ConsumerStopReq struct {
	Type         FrameType `json:"type"`
	SubscribeIDs []string  `json:"subscribe_ids"`
}

// ConsumerStopResp reports which requested subscriptions had a live local
// consumer. Missing IDs are not errors because a server subscription can be
// cancelled after its foreground consumer has already exited.
type ConsumerStopResp struct {
	Type     FrameType `json:"type"`
	Stopped  []string  `json:"stopped,omitempty"`
	NotFound []string  `json:"not_found,omitempty"`
	Error    string    `json:"error,omitempty"`
}

// StatusReq is an empty JSON frame ad-hoc tooling sends after Hello to
// request a full StatusResp. Bus replies with one StatusResp then closes
// the connection.
type StatusReq struct {
	Type FrameType `json:"type"`
}

// StatusResp is the bus's snapshot view for `dws event status` rendering.
// Counts are accumulated since bus start; per-consumer entries are sorted
// by PID for deterministic output.
type StatusResp struct {
	Type                 FrameType           `json:"type"`
	Bus                  StatusBus           `json:"bus"`
	SourceState          StatusSource        `json:"source_state"`
	Consumers            []StatusConsumer    `json:"consumers"`
	PerEventTypeCounters map[string]Counters `json:"per_event_type"`
}

// StatusBus is the bus daemon's identity / lifecycle view.
type StatusBus struct {
	PID            int                 `json:"pid"`
	UptimeSecs     int64               `json:"uptime_secs"`
	IdleTimeoutSec int                 `json:"idle_timeout_secs"`
	ClientID       string              `json:"client_id"`
	Edition        string              `json:"edition"`
	SourceKind     dwsevent.SourceKind `json:"source_kind,omitempty"`
	IdentityHash   string              `json:"identity_hash,omitempty"`
	SourceID       string              `json:"source_id,omitempty"`
}

// StatusSource is the source.Machine snapshot at status RPC time.
type StatusSource struct {
	State           string `json:"state"`
	Source          string `json:"source"`
	LastEventAtMS   int64  `json:"last_event_at_ms,omitempty"`
	LastReconnectMS int64  `json:"last_reconnect_at_ms,omitempty"`
	ReconnectCount  int    `json:"reconnect_count"`
}

// StatusConsumer is one consumer's per-IPC-connection view.
type StatusConsumer struct {
	PID            int      `json:"pid"`
	EventTypes     []string `json:"event_types,omitempty"`
	Filter         string   `json:"filter,omitempty"`
	SubscribeID    string   `json:"subscribe_id,omitempty"`
	SubscribedAtMS int64    `json:"subscribed_at_ms"`
	Received       uint64   `json:"received"`
	Dropped        uint64   `json:"dropped"`
}

// Counters is the shared "received / dropped" pair used by per-consumer
// and per-event-type rollups.
type Counters struct {
	Received uint64 `json:"received"`
	Dropped  uint64 `json:"dropped"`
}

// typeOnly is the minimal envelope used by DecodeFrame to peek at the
// "type" field before deciding which concrete struct to unmarshal into.
type typeOnly struct {
	Type FrameType `json:"type"`
}

// PeekType returns the FrameType of a raw frame without fully decoding it.
// Returns an error when the JSON is malformed or the type field is missing.
func PeekType(raw []byte) (FrameType, error) {
	var t typeOnly
	if err := json.Unmarshal(raw, &t); err != nil {
		return "", fmt.Errorf("transport: peek frame type: %w", err)
	}
	if t.Type == "" {
		return "", fmt.Errorf("transport: frame missing 'type' field")
	}
	return t.Type, nil
}