larksuite--cli/internal/event/consume/handshake.go

// Copyright (c) 2026 Lark Technologies Pte. Ltd.
// SPDX-License-Identifier: MIT

package consume

import (
	"bufio"
	"bytes"
	"fmt"
	"net"
	"os"
	"time"

	"github.com/larksuite/cli/internal/event/protocol"
)

const helloAckTimeout = 5 * time.Second // symmetric with bus-side hello read deadline

// doHello returns a bufio.Reader holding any bytes already pulled off conn so events
// buffered with the ack in one TCP segment aren't dropped.
func doHello(conn net.Conn, eventKey string, eventTypes []string, subscriptionID string) (*protocol.HelloAck, *bufio.Reader, error) {
	hello := protocol.NewHello(os.Getpid(), eventKey, eventTypes, "v1", subscriptionID)
	if err := protocol.EncodeWithDeadline(conn, hello, protocol.WriteTimeout); err != nil {
		return nil, nil, err
	}

	if err := conn.SetReadDeadline(time.Now().Add(helloAckTimeout)); err != nil {
		return nil, nil, fmt.Errorf("set hello_ack deadline: %w", err)
	}
	br := bufio.NewReader(conn)
	line, err := protocol.ReadFrame(br)
	if err != nil {
		return nil, nil, fmt.Errorf("no hello_ack received: %w", err)
	}
	// best-effort clear; if the conn is already broken, the loop's first read will surface it
	_ = conn.SetReadDeadline(time.Time{})
	msg, err := protocol.Decode(bytes.TrimRight(line, "\n"))
	if err != nil {
		return nil, nil, err
	}
	ack, ok := msg.(*protocol.HelloAck)
	if !ok {
		return nil, nil, fmt.Errorf("expected hello_ack, got %T", msg)
	}
	return ack, br, nil
}