chore: import upstream snapshot with attribution

internal/event/protocol/codec.go

@@ -0,0 +1,106 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+// Package protocol defines the newline-delimited JSON wire format used over IPC.
+package protocol
+
+import (
+	"bufio"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"time"
+)
+
+const MaxFrameBytes = 1 << 20 // reject larger frames to bound reader buffer growth
+
+// ErrFrameTooLarge is returned by ReadFrame when a single frame exceeds MaxFrameBytes.
+var ErrFrameTooLarge = errors.New("protocol: frame exceeds MaxFrameBytes")
+
+const WriteTimeout = 5 * time.Second // bound writes against wedged peer kernel buffer
+
+type typeEnvelope struct {
+	Type string `json:"type"`
+}
+
+func Encode(w io.Writer, msg interface{}) error {
+	data, err := json.Marshal(msg)
+	if err != nil {
+		return fmt.Errorf("protocol encode: %w", err)
+	}
+	data = append(data, '\n')
+	_, err = w.Write(data)
+	return err
+}
+
+func EncodeWithDeadline(conn net.Conn, msg interface{}, timeout time.Duration) error {
+	if err := conn.SetWriteDeadline(time.Now().Add(timeout)); err != nil {
+		return err
+	}
+	return Encode(conn, msg)
+}
+
+// ReadFrame reads one newline-delimited message; caps at MaxFrameBytes to defang slowloris.
+func ReadFrame(br *bufio.Reader) ([]byte, error) {
+	var buf []byte
+	for {
+		chunk, err := br.ReadSlice('\n')
+		switch err {
+		case nil:
+			if len(buf) == 0 {
+				return chunk, nil
+			}
+			if len(buf)+len(chunk) > MaxFrameBytes {
+				return nil, ErrFrameTooLarge
+			}
+			return append(buf, chunk...), nil
+		case bufio.ErrBufferFull:
+			if len(buf)+len(chunk) > MaxFrameBytes {
+				return nil, ErrFrameTooLarge
+			}
+			buf = append(buf, chunk...)
+		default:
+			return nil, err
+		}
+	}
+}
+
+func Decode(line []byte) (interface{}, error) {
+	var env typeEnvelope
+	if err := json.Unmarshal(line, &env); err != nil {
+		return nil, fmt.Errorf("protocol decode type: %w", err)
+	}
+
+	var msg interface{}
+	switch env.Type {
+	case MsgTypeHello:
+		msg = &Hello{}
+	case MsgTypeHelloAck:
+		msg = &HelloAck{}
+	case MsgTypeEvent:
+		msg = &Event{}
+	case MsgTypeBye:
+		msg = &Bye{}
+	case MsgTypePreShutdownCheck:
+		msg = &PreShutdownCheck{}
+	case MsgTypePreShutdownAck:
+		msg = &PreShutdownAck{}
+	case MsgTypeStatusQuery:
+		msg = &StatusQuery{}
+	case MsgTypeStatusResponse:
+		msg = &StatusResponse{}
+	case MsgTypeShutdown:
+		msg = &Shutdown{}
+	case MsgTypeSourceStatus:
+		msg = &SourceStatus{}
+	default:
+		return nil, fmt.Errorf("protocol: unknown message type %q", env.Type)
+	}
+
+	if err := json.Unmarshal(line, msg); err != nil {
+		return nil, fmt.Errorf("protocol decode %s: %w", env.Type, err)
+	}
+	return msg, nil
+}

internal/event/protocol/codec_test.go

@@ -0,0 +1,164 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package protocol
+
+import (
+	"bytes"
+	"encoding/json"
+	"testing"
+)
+
+func TestEncodeDecodeHello(t *testing.T) {
+	msg := &Hello{
+		Type:       MsgTypeHello,
+		PID:        12345,
+		EventKey:   "mail.user_mailbox.event.message_received_v1",
+		EventTypes: []string{"mail.user_mailbox.event.message_received_v1"},
+		Version:    "v1",
+	}
+
+	var buf bytes.Buffer
+	if err := Encode(&buf, msg); err != nil {
+		t.Fatalf("encode: %v", err)
+	}
+
+	decoded, err := Decode(buf.Bytes())
+	if err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	hello, ok := decoded.(*Hello)
+	if !ok {
+		t.Fatalf("expected *Hello, got %T", decoded)
+	}
+	if hello.PID != 12345 || hello.EventKey != "mail.user_mailbox.event.message_received_v1" {
+		t.Errorf("unexpected hello: %+v", hello)
+	}
+}
+
+func TestEncodeDecodeEvent(t *testing.T) {
+	payload := json.RawMessage(`{"foo":"bar"}`)
+	msg := &Event{
+		Type:      MsgTypeEvent,
+		EventType: "im.message.receive_v1",
+		Payload:   payload,
+	}
+
+	var buf bytes.Buffer
+	if err := Encode(&buf, msg); err != nil {
+		t.Fatalf("encode: %v", err)
+	}
+
+	decoded, err := Decode(buf.Bytes())
+	if err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	evt, ok := decoded.(*Event)
+	if !ok {
+		t.Fatalf("expected *Event, got %T", decoded)
+	}
+	if evt.EventType != "im.message.receive_v1" {
+		t.Errorf("got event_type %q", evt.EventType)
+	}
+}
+
+func TestEncodeAddsNewline(t *testing.T) {
+	msg := &Bye{Type: MsgTypeBye}
+	var buf bytes.Buffer
+	Encode(&buf, msg)
+	if buf.Bytes()[buf.Len()-1] != '\n' {
+		t.Error("encoded message should end with newline")
+	}
+}
+
+func TestDecodeUnknownType(t *testing.T) {
+	_, err := Decode([]byte(`{"type":"unknown_xyz"}`))
+	if err == nil {
+		t.Error("expected error for unknown type")
+	}
+}
+
+func TestEncodeDecodeHello_WithSubscriptionID(t *testing.T) {
+	msg := &Hello{
+		Type:           MsgTypeHello,
+		PID:            12345,
+		EventKey:       "mail.user_mailbox.event.message_received_v1",
+		EventTypes:     []string{"mail.user_mailbox.event.message_received_v1"},
+		Version:        "v1",
+		SubscriptionID: "mail.user_mailbox.event.message_received_v1:a7Bx9Kp2Lm3Qv4Rs",
+	}
+	buf := &bytes.Buffer{}
+	if err := Encode(buf, msg); err != nil {
+		t.Fatal(err)
+	}
+	line := buf.Bytes()
+	if !bytes.Contains(line, []byte(`"subscription_id":"mail.user_mailbox.event.message_received_v1:a7Bx9Kp2Lm3Qv4Rs"`)) {
+		t.Errorf("subscription_id not serialized: %s", string(line))
+	}
+	decoded, err := Decode(bytes.TrimRight(line, "\n"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	hello, ok := decoded.(*Hello)
+	if !ok {
+		t.Fatalf("expected *Hello, got %T", decoded)
+	}
+	if hello.SubscriptionID != msg.SubscriptionID {
+		t.Errorf("roundtrip subscription_id: got %q want %q", hello.SubscriptionID, msg.SubscriptionID)
+	}
+}
+
+func TestEncodeDecodeHello_EmptySubscriptionIDOmitted(t *testing.T) {
+	msg := &Hello{
+		Type:       MsgTypeHello,
+		PID:        1,
+		EventKey:   "k",
+		EventTypes: []string{"k"},
+		Version:    "v1",
+	}
+	buf := &bytes.Buffer{}
+	if err := Encode(buf, msg); err != nil {
+		t.Fatal(err)
+	}
+	if bytes.Contains(buf.Bytes(), []byte("subscription_id")) {
+		t.Errorf("empty subscription_id should be omitted: %s", buf.String())
+	}
+	decoded, _ := Decode(bytes.TrimRight(buf.Bytes(), "\n"))
+	hello := decoded.(*Hello)
+	if hello.SubscriptionID != "" {
+		t.Errorf("got %q, want empty", hello.SubscriptionID)
+	}
+}
+
+func TestEncodeDecodePreShutdownCheck_WithSubscriptionID(t *testing.T) {
+	msg := &PreShutdownCheck{
+		Type:           MsgTypePreShutdownCheck,
+		EventKey:       "mail.x",
+		SubscriptionID: "mail.x:abc",
+	}
+	buf := &bytes.Buffer{}
+	if err := Encode(buf, msg); err != nil {
+		t.Fatal(err)
+	}
+	decoded, err := Decode(bytes.TrimRight(buf.Bytes(), "\n"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := decoded.(*PreShutdownCheck)
+	if got.SubscriptionID != msg.SubscriptionID {
+		t.Errorf("roundtrip: got %q want %q", got.SubscriptionID, msg.SubscriptionID)
+	}
+}
+
+func TestStatusResponse_ConsumerInfo_SubscriptionID(t *testing.T) {
+	msg := NewStatusResponse(7, 120, 1, []ConsumerInfo{
+		{PID: 99, EventKey: "mail.x", SubscriptionID: "mail.x:abc", Received: 5, Dropped: 0},
+	})
+	buf := &bytes.Buffer{}
+	if err := Encode(buf, msg); err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Contains(buf.Bytes(), []byte(`"subscription_id":"mail.x:abc"`)) {
+		t.Errorf("ConsumerInfo.SubscriptionID missing from JSON: %s", buf.String())
+	}
+}

internal/event/protocol/messages.go

@@ -0,0 +1,175 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package protocol
+
+import "encoding/json"
+
+const (
+	MsgTypeHello            = "hello"
+	MsgTypeHelloAck         = "hello_ack"
+	MsgTypeEvent            = "event"
+	MsgTypeBye              = "bye"
+	MsgTypePreShutdownCheck = "pre_shutdown_check"
+	MsgTypePreShutdownAck   = "pre_shutdown_ack"
+	MsgTypeStatusQuery      = "status_query"
+	MsgTypeStatusResponse   = "status_response"
+	MsgTypeShutdown         = "shutdown"
+	MsgTypeSourceStatus     = "source_status"
+)
+
+const (
+	SourceStateConnecting   = "connecting"
+	SourceStateConnected    = "connected"
+	SourceStateDisconnected = "disconnected"
+	SourceStateReconnecting = "reconnecting"
+)
+
+// SourceStatus is best-effort: hub drops it when consumer's send channel is full.
+type SourceStatus struct {
+	Type   string `json:"type"`
+	Source string `json:"source"`
+	State  string `json:"state"`
+	Detail string `json:"detail,omitempty"`
+}
+
+type Hello struct {
+	Type           string   `json:"type"`
+	PID            int      `json:"pid"`
+	EventKey       string   `json:"event_key"`
+	EventTypes     []string `json:"event_types"`
+	Version        string   `json:"version"`
+	SubscriptionID string   `json:"subscription_id,omitempty"` // empty = fallback to EventKey on bus side
+}
+
+type HelloAck struct {
+	Type         string `json:"type"`
+	BusVersion   string `json:"bus_version"`
+	FirstForKey  bool   `json:"first_for_key"`
+	Rejected     bool   `json:"rejected,omitempty"`
+	RejectReason string `json:"reject_reason,omitempty"`
+}
+
+// Event: Seq is per-conn monotonic; gaps signal bus drop-oldest backpressure loss.
+type Event struct {
+	Type       string          `json:"type"`
+	EventType  string          `json:"event_type"`
+	EventID    string          `json:"event_id,omitempty"`
+	SourceTime string          `json:"source_time,omitempty"` // ms-precision unix timestamp, stringified
+	Seq        uint64          `json:"seq,omitempty"`
+	Payload    json.RawMessage `json:"payload"`
+}
+
+type Bye struct {
+	Type string `json:"type"`
+}
+
+// PreShutdownCheck atomically reserves the cleanup lock for (EventKey, SubscriptionID).
+type PreShutdownCheck struct {
+	Type           string `json:"type"`
+	EventKey       string `json:"event_key"`
+	SubscriptionID string `json:"subscription_id,omitempty"` // empty = fallback to EventKey
+}
+
+type PreShutdownAck struct {
+	Type       string `json:"type"`
+	LastForKey bool   `json:"last_for_key"`
+}
+
+type StatusQuery struct {
+	Type string `json:"type"`
+}
+
+type ConsumerInfo struct {
+	PID            int    `json:"pid"`
+	EventKey       string `json:"event_key"`
+	SubscriptionID string `json:"subscription_id,omitempty"`
+	Received       int64  `json:"received"`
+	Dropped        int64  `json:"dropped"`
+}
+
+type StatusResponse struct {
+	Type        string         `json:"type"`
+	PID         int            `json:"pid"`
+	UptimeSec   int            `json:"uptime_sec"`
+	ActiveConns int            `json:"active_conns"`
+	Consumers   []ConsumerInfo `json:"consumers"`
+}
+
+type Shutdown struct {
+	Type string `json:"type"`
+}
+
+func NewHello(pid int, eventKey string, eventTypes []string, version string, subscriptionID string) *Hello {
+	return &Hello{
+		Type:           MsgTypeHello,
+		PID:            pid,
+		EventKey:       eventKey,
+		EventTypes:     eventTypes,
+		Version:        version,
+		SubscriptionID: subscriptionID,
+	}
+}
+
+func NewHelloAck(busVersion string, firstForKey bool) *HelloAck {
+	return &HelloAck{
+		Type:        MsgTypeHelloAck,
+		BusVersion:  busVersion,
+		FirstForKey: firstForKey,
+	}
+}
+
+// NewHelloAckRejected builds a hello_ack that tells the consumer the bus refused
+// registration (e.g. a SingleConsumer EventKey already has a running consumer).
+func NewHelloAckRejected(busVersion, reason string) *HelloAck {
+	return &HelloAck{
+		Type:         MsgTypeHelloAck,
+		BusVersion:   busVersion,
+		Rejected:     true,
+		RejectReason: reason,
+	}
+}
+
+func NewEvent(eventType, eventID, sourceTime string, seq uint64, payload json.RawMessage) *Event {
+	return &Event{
+		Type:       MsgTypeEvent,
+		EventType:  eventType,
+		EventID:    eventID,
+		SourceTime: sourceTime,
+		Seq:        seq,
+		Payload:    payload,
+	}
+}
+
+func NewPreShutdownCheck(eventKey, subscriptionID string) *PreShutdownCheck {
+	return &PreShutdownCheck{Type: MsgTypePreShutdownCheck, EventKey: eventKey, SubscriptionID: subscriptionID}
+}
+
+func NewPreShutdownAck(lastForKey bool) *PreShutdownAck {
+	return &PreShutdownAck{Type: MsgTypePreShutdownAck, LastForKey: lastForKey}
+}
+
+func NewStatusQuery() *StatusQuery {
+	return &StatusQuery{Type: MsgTypeStatusQuery}
+}
+
+func NewStatusResponse(pid int, uptimeSec int, activeConns int, consumers []ConsumerInfo) *StatusResponse {
+	return &StatusResponse{
+		Type:        MsgTypeStatusResponse,
+		PID:         pid,
+		UptimeSec:   uptimeSec,
+		ActiveConns: activeConns,
+		Consumers:   consumers,
+	}
+}
+
+func NewShutdown() *Shutdown { return &Shutdown{Type: MsgTypeShutdown} }
+
+func NewSourceStatus(source, state, detail string) *SourceStatus {
+	return &SourceStatus{
+		Type:   MsgTypeSourceStatus,
+		Source: source,
+		State:  state,
+		Detail: detail,
+	}
+}

internal/event/protocol/messages_test.go

@@ -0,0 +1,129 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package protocol
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"testing"
+	"time"
+)
+
+// Every NewXxx helper must set the Type discriminator (Decode rejects messages without it).
+func TestConstructors_PinTypeField(t *testing.T) {
+	if got := NewHello(1, "k", []string{"t"}, "v1", ""); got.Type != MsgTypeHello {
+		t.Errorf("NewHello.Type = %q, want %q", got.Type, MsgTypeHello)
+	}
+	if got := NewHelloAck("v1", true); got.Type != MsgTypeHelloAck || !got.FirstForKey {
+		t.Errorf("NewHelloAck mismatch: %+v", got)
+	}
+	if got := NewEvent("im.msg", "e1", "", 7, json.RawMessage(`{}`)); got.Type != MsgTypeEvent || got.Seq != 7 {
+		t.Errorf("NewEvent mismatch: %+v", got)
+	}
+	if got := NewPreShutdownCheck("k", ""); got.Type != MsgTypePreShutdownCheck || got.EventKey != "k" {
+		t.Errorf("NewPreShutdownCheck mismatch: %+v", got)
+	}
+	if got := NewPreShutdownAck(true); got.Type != MsgTypePreShutdownAck || !got.LastForKey {
+		t.Errorf("NewPreShutdownAck mismatch: %+v", got)
+	}
+	if got := NewStatusQuery(); got.Type != MsgTypeStatusQuery {
+		t.Errorf("NewStatusQuery.Type = %q", got.Type)
+	}
+	if got := NewStatusResponse(42, 10, 2, []ConsumerInfo{{PID: 1}, {PID: 2}}); got.Type != MsgTypeStatusResponse || got.PID != 42 || len(got.Consumers) != 2 {
+		t.Errorf("NewStatusResponse mismatch: %+v", got)
+	}
+	if got := NewShutdown(); got.Type != MsgTypeShutdown {
+		t.Errorf("NewShutdown.Type = %q", got.Type)
+	}
+	if got := NewSourceStatus("feishu-ws", SourceStateConnected, "ok"); got.Type != MsgTypeSourceStatus || got.Detail != "ok" {
+		t.Errorf("NewSourceStatus mismatch: %+v", got)
+	}
+}
+
+func TestEncode_DecodeRoundtripAllTypes(t *testing.T) {
+	roundtrip := func(t *testing.T, msg interface{}, want interface{}) {
+		t.Helper()
+		var buf bytes.Buffer
+		if err := Encode(&buf, msg); err != nil {
+			t.Fatalf("encode: %v", err)
+		}
+		line := bytes.TrimRight(buf.Bytes(), "\n")
+		got, err := Decode(line)
+		if err != nil {
+			t.Fatalf("decode: %v", err)
+		}
+		if gotT, wantT := fmt.Sprintf("%T", got), fmt.Sprintf("%T", want); gotT != wantT {
+			t.Errorf("decoded type = %s, want %s", gotT, wantT)
+		}
+	}
+	roundtrip(t, NewHelloAck("v1", true), &HelloAck{})
+	roundtrip(t, NewPreShutdownCheck("im.msg", ""), &PreShutdownCheck{})
+	roundtrip(t, NewPreShutdownAck(false), &PreShutdownAck{})
+	roundtrip(t, NewStatusQuery(), &StatusQuery{})
+	roundtrip(t, NewStatusResponse(7, 120, 1, []ConsumerInfo{{PID: 99, EventKey: "k"}}), &StatusResponse{})
+	roundtrip(t, NewShutdown(), &Shutdown{})
+	roundtrip(t, NewSourceStatus("feishu", SourceStateReconnecting, "attempt 2"), &SourceStatus{})
+	roundtrip(t, &Bye{Type: MsgTypeBye}, &Bye{})
+}
+
+// EncodeWithDeadline must apply a write deadline so a wedged peer can't stall the writer forever.
+func TestEncodeWithDeadline_AppliesDeadline(t *testing.T) {
+	client, server := net.Pipe()
+	defer client.Close()
+	defer server.Close()
+	start := time.Now()
+	err := EncodeWithDeadline(client, NewShutdown(), 100*time.Millisecond)
+	elapsed := time.Since(start)
+	if err == nil {
+		t.Fatal("expected deadline error, got nil")
+	}
+	if elapsed > 500*time.Millisecond {
+		t.Errorf("EncodeWithDeadline didn't honour deadline: took %v (want ~100ms)", elapsed)
+	}
+}
+
+func TestReadFrame_RejectsOversized(t *testing.T) {
+	big := bytes.Repeat([]byte("a"), MaxFrameBytes+1)
+	big = append(big, '\n')
+	br := bufio.NewReader(bytes.NewReader(big))
+	_, err := ReadFrame(br)
+	if !errors.Is(err, ErrFrameTooLarge) {
+		t.Fatalf("ReadFrame on oversized input: err = %v, want ErrFrameTooLarge", err)
+	}
+}
+
+func TestReadFrame_PropagatesEOF(t *testing.T) {
+	br := bufio.NewReader(bytes.NewReader(nil))
+	_, err := ReadFrame(br)
+	if err != io.EOF {
+		t.Errorf("err = %v, want io.EOF", err)
+	}
+}
+
+func TestHelloAckRejected_RoundTrip(t *testing.T) {
+	ack := NewHelloAckRejected("v1", "another consumer (pid 42) is already running for this subscription")
+	if !ack.Rejected || ack.RejectReason == "" {
+		t.Fatalf("NewHelloAckRejected fields: %+v", ack)
+	}
+	var buf bytes.Buffer
+	if err := Encode(&buf, ack); err != nil {
+		t.Fatalf("encode: %v", err)
+	}
+	msg, err := Decode(bytes.TrimRight(buf.Bytes(), "\n"))
+	if err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	got, ok := msg.(*HelloAck)
+	if !ok {
+		t.Fatalf("decoded type = %T, want *HelloAck", msg)
+	}
+	if !got.Rejected || got.RejectReason != ack.RejectReason {
+		t.Errorf("roundtrip = %+v, want Rejected with reason", got)
+	}
+}