feat(integrations/lark): align WS to official binary Frame protocol + DispatchResult outbound replies (MUL-2671)

Two must-fix items from Elon's review of PR #3277:

1. WS protocol layer rewritten to match the official Lark Go SDK
   (`larksuite/oapi-sdk-go/v3/ws`):
   - Bootstrap is `POST /callback/ws/endpoint` with AppID/AppSecret
     in the body (no tenant_access_token bearer). Response carries
     wss URL + ClientConfig (PingInterval / ReconnectInterval /
     ReconnectNonce / ReconnectCount).
   - `service_id` is parsed from the wss URL query and used as
     Frame.Service on every outbound frame.
   - Wire envelope is the binary protobuf `pbbp2.Frame` (hand-rolled
     via protowire to avoid pulling the whole SDK in, byte-identical
     field tags). JSON payloads are nested inside Frame.Payload.
   - Inbound data frames are ACKed with a `Response{code:200,...}`
     JSON payload that reuses the inbound headers; infra failures
     produce code=500 so Lark retries.
   - Ping is the app-layer binary `NewPingFrame(serviceID)` at the
     server-supplied cadence; WebSocket protocol PING is removed
     (Lark ignores it). Server-initiated pings get a pong reply.
   - ctx-cancel-breaks-read invariant preserved via the watchdog
     goroutine that closes the conn on ctx.Done; the read loop and
     ping goroutine serialize their writes through a single mutex.

2. `DispatchResult` outbound replies wired via a new `OutcomeReplier`:
   - `OutcomeNeedsBinding` mints a one-shot binding token and sends
     the binding prompt card to the sender's open_id.
   - `OutcomeAgentOffline` / `OutcomeAgentArchived` push a notice
     card into the chat with the agent name + Chinese copy matching
     §4.6.
   - `OutcomeIngested` stays owned by the Patcher; `OutcomeDropped`
     is silent.
   - The replier is best-effort: outbound failures are logged and
     swallowed so a Lark outage cannot stall the inbound pipeline.
   - Hub installs the noop replier by default; router wires the
     production `LarkOutcomeReplier` when APIClient.IsConfigured().

PersonalAgent long-conn risk surfaced (open per Feishu docs:
`长连接模式仅支持企业自建应用`). The implementation works for any
app archetype; the open question is whether `/callback/ws/endpoint`
accepts PersonalAgent credentials in practice. Surfacing the Lark
code+msg verbatim from the bootstrap response so an operator running
the smoke test sees the exact failure rather than a generic timeout.

Co-authored-by: multica-agent <[email protected]>
This commit is contained in:
J
2026-05-28 12:58:26 +08:00
co-authored by multica-agent
parent 48f779e5e7
commit 9540008a1a
12 changed files with 1852 additions and 535 deletions
+34 -15
View File
@@ -254,6 +254,25 @@ func NewRouterWithOptions(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus
// piggybacks on its tenant_access_token cache.
connectorFactory, connectorLabel := buildLarkConnectorFactory(larkClient, installSvc)
h.LarkHub = lark.NewHub(queries, connectorFactory, dispatcher, lark.HubConfig{})
// OutcomeReplier wires the outbound side of the
// EventEmitter contract: NeedsBinding / AgentOffline /
// AgentArchived translate to a Lark-side reply card.
// Requires the real APIClient (the stub returns
// ErrAPIClientNotConfigured on every send) and the
// binding token service. When either is missing, the
// Hub falls back to the noop replier and the outcomes
// get logged but not delivered — clearly visible in
// boot output so operators understand the gap.
replier := lark.NewLarkOutcomeReplier(lark.OutcomeReplierConfig{
APIClient: larkClient,
BindingSvc: h.LarkBindingTokens,
Credentials: installSvc,
Queries: queries,
PublicURL: signupConfig.PublicURL,
Logger: slog.Default(),
})
h.LarkHub.SetOutcomeReplier(replier)
slog.Info("lark inbound pipeline wired", "connector", connectorLabel)
// Device-flow registration service: end-to-end install
@@ -861,28 +880,27 @@ func NewRouterWithOptions(pool *pgxpool.Pool, hub *realtime.Hub, bus *events.Bus
// buildLarkConnectorFactory picks between the staging-mode
// NoopConnector and the real WS long-conn connector based on
// MULTICA_LARK_WS_ENABLED. The real connector requires an HTTP
// APIClient that can mint a tenant_access_token; the stub cannot, so we
// fall back to noop with a warning if the operator enables the WS flag
// without also enabling MULTICA_LARK_HTTP_ENABLED. Returns the factory
// the Hub will use plus a short label for the boot log so operators
// can see at a glance which mode they are in.
// MULTICA_LARK_WS_ENABLED. The real connector talks to
// /callback/ws/endpoint directly with app_id/app_secret (no
// tenant_access_token bearer needed for the bootstrap), so it can
// run independently of MULTICA_LARK_HTTP_ENABLED — but outbound
// reply cards (binding prompt, offline / archived notices) require
// the HTTP APIClient. We still recommend enabling both together;
// when only WS is on, the OutcomeReplier surfaces a warning and
// downgrades to silent drops.
//
// Returns the factory plus a short label for the boot log so
// operators can see at a glance which mode they are in.
func buildLarkConnectorFactory(client lark.APIClient, installSvc *lark.InstallationService) (lark.ConnectorFactory, string) {
if !strings.EqualFold(strings.TrimSpace(os.Getenv("MULTICA_LARK_WS_ENABLED")), "true") {
return lark.NoopConnectorFactory(slog.Default()), "noop"
}
tokenSource, err := lark.NewHTTPAPIClientTokenSource(client)
if err != nil {
slog.Warn("lark ws: MULTICA_LARK_WS_ENABLED requires MULTICA_LARK_HTTP_ENABLED; falling back to noop", "error", err)
return lark.NoopConnectorFactory(slog.Default()), "noop"
}
endpointFetcher, err := lark.NewHTTPConnectionTokenFetcher(lark.HTTPConnectionTokenConfig{
BaseURL: strings.TrimSpace(os.Getenv("MULTICA_LARK_HTTP_BASE_URL")),
TokenSource: tokenSource,
Logger: slog.Default(),
BaseURL: strings.TrimSpace(os.Getenv("MULTICA_LARK_CALLBACK_BASE_URL")),
Logger: slog.Default(),
})
if err != nil {
slog.Error("lark ws: connection_token fetcher init failed; falling back to noop", "error", err)
slog.Error("lark ws: endpoint fetcher init failed; falling back to noop", "error", err)
return lark.NoopConnectorFactory(slog.Default()), "noop"
}
decoder := lark.NewLarkJSONFrameDecoder()
@@ -912,6 +930,7 @@ func buildLarkConnectorFactory(client lark.APIClient, installSvc *lark.Installat
slog.Error("lark ws: connector init failed; falling back to noop", "error", err)
return lark.NoopConnectorFactory(slog.Default()), "noop"
}
_ = client // outbound APIClient flows in through OutcomeReplier wiring on the Hub
return func(_ db.LarkInstallation) (lark.EventConnector, error) {
return conn, nil
}, "ws-long-conn"
+14 -5
View File
@@ -15,18 +15,27 @@
// for offline / archived; emit returns DispatchResult + error so
// the connector can post the matching Lark-side reply card)
// 6. AuditLogger (lark_inbound_audit; deliberately no body column)
// 7. APIClient interface + stub (transport surface for outbound;
// real Lark wire-protocol implementation lands in a follow-up
// behind this same interface)
// 7. APIClient interface + http_client.go (real Lark Open Platform
// transport for IM v1 send/patch + binding prompt + bot info;
// stubAPIClient refuses calls when no production client is wired)
// 8. Hub (WS lease + per-installation supervisor goroutines with
// exponential backoff + jitter; renewer cancels the connector's
// run ctx on lease loss to keep §4.4 ownership safe across
// replicas; EventConnector interface is the seam for the real
// wire protocol)
// 9. Patcher (subscribes to task / chat-done events; keeps the
// 9. WSLongConnConnector (real long-conn over gorilla/websocket; the
// wire protocol is the binary Frame envelope from the official Go
// SDK — bootstrap via POST /callback/ws/endpoint, app-layer
// ping/pong, ACK responses on every data frame, ctx cancel breaks
// blocking ReadMessage via a watchdog goroutine for §4.4)
// 10. Patcher (subscribes to task / chat-done events; keeps the
// per-task Lark interactive card in sync; throttled patches +
// final/error bypass)
// 10. RegistrationService (RFC 8628 device-flow scan-to-install: opens
// 11. OutcomeReplier (outbound side of the EventEmitter contract:
// NeedsBinding mints a token + sends the binding prompt;
// AgentOffline / AgentArchived push status notice cards into the
// chat; Ingested is owned by the Patcher; Dropped is silent)
// 12. RegistrationService (RFC 8628 device-flow scan-to-install: opens
// a session against accounts.feishu.cn, polls in the background,
// and on success writes through InstallationService + auto-binds
// the installer via InstallerBinder so §2.1 "scan to bind, you're
+34 -11
View File
@@ -189,6 +189,7 @@ type Hub struct {
queries HubQueries
factory ConnectorFactory
dispatcher *Dispatcher
replier OutcomeReplier
cfg HubConfig
// nodeID is the per-process lease ownership token. The CAS
@@ -206,13 +207,19 @@ type Hub struct {
// NewHub constructs a Hub bound to the supplied queries, connector
// factory and dispatcher. The Hub does not start any goroutines until
// Run is called.
// Run is called. The replier (OutcomeReplier) handles the outbound
// side of the EventEmitter contract — NeedsBinding / AgentOffline /
// AgentArchived cards — and is best-effort: failures are logged and
// do not interrupt inbound processing. A nil replier falls back to
// the noop replier so callers that have not wired outbound replies
// yet still get the inbound pipeline running.
func NewHub(queries HubQueries, factory ConnectorFactory, dispatcher *Dispatcher, cfg HubConfig) *Hub {
cfg = cfg.withDefaults()
return &Hub{
queries: queries,
factory: factory,
dispatcher: dispatcher,
replier: NewNoopOutcomeReplier(cfg.Logger),
cfg: cfg,
nodeID: newNodeID(),
stopFns: make(map[string]context.CancelFunc),
@@ -220,6 +227,17 @@ func NewHub(queries HubQueries, factory ConnectorFactory, dispatcher *Dispatcher
}
}
// SetOutcomeReplier installs the production replier on the Hub. Must
// be called BEFORE Run; setting it afterwards is a data race against
// the supervisor's emit goroutines. Nil resets back to the noop
// replier (useful for tests).
func (h *Hub) SetOutcomeReplier(r OutcomeReplier) {
if r == nil {
r = NewNoopOutcomeReplier(h.cfg.Logger)
}
h.replier = r
}
// NodeID exposes the per-process lease token. Useful for tests and
// for observability (so operators can correlate DB lease rows to a
// running replica).
@@ -414,7 +432,7 @@ func (h *Hub) supervise(ctx context.Context, inst db.LarkInstallation, id string
startedAt := h.cfg.Now()
runErr := conn.Run(runCtx, inst, func(emitCtx context.Context, msg InboundMessage) (DispatchResult, error) {
return h.handleEvent(emitCtx, log, msg)
return h.handleEvent(emitCtx, inst, log, msg)
})
runCancel()
<-renewDone
@@ -530,16 +548,18 @@ func (h *Hub) releaseLease(instID pgtype.UUID) {
}
// handleEvent is the seam between the connector (which emits normalized
// InboundMessage) and the inbound Dispatcher. We deliberately do not
// retry here — the Dispatcher classifies errors itself (productizable
// outcomes vs. infra failures), and infra failures propagate up to the
// connector, which decides whether to reconnect.
// InboundMessage) and the inbound Dispatcher + outbound OutcomeReplier.
// We deliberately do not retry here — the Dispatcher classifies errors
// itself (productizable outcomes vs. infra failures), and infra
// failures propagate up to the connector, which decides whether to
// reconnect.
//
// The result + error are returned to the connector via the EventEmitter
// callback so it can post the Lark-side reply that matches the
// outcome (binding card, offline card, etc.) and react to infra
// failures (e.g. tear the WS down so the Hub re-establishes it).
func (h *Hub) handleEvent(ctx context.Context, log *slog.Logger, msg InboundMessage) (DispatchResult, error) {
// On a clean dispatch (err==nil) we invoke the replier inline so the
// connector returns to its read loop only after the Lark-side reply
// has been attempted. The replier is best-effort — its own failures
// are logged and swallowed so an outbound Lark outage cannot stall
// the inbound pipeline.
func (h *Hub) handleEvent(ctx context.Context, inst db.LarkInstallation, log *slog.Logger, msg InboundMessage) (DispatchResult, error) {
if h.dispatcher == nil {
log.Warn("lark hub: dispatcher not configured; dropping event",
"event_id", msg.EventID,
@@ -559,6 +579,9 @@ func (h *Hub) handleEvent(ctx context.Context, log *slog.Logger, msg InboundMess
"outcome", string(res.Outcome),
"drop_reason", string(res.DropReason),
)
if h.replier != nil {
h.replier.Reply(ctx, inst, msg, res)
}
return res, nil
}
@@ -0,0 +1,281 @@
package lark
import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"net/url"
"strings"
"github.com/jackc/pgx/v5/pgtype"
db "github.com/multica-ai/multica/server/pkg/db/generated"
)
// OutcomeReplier reacts to the Dispatcher's verdict by posting the
// appropriate Lark-side reply card. This is the outbound half of the
// `EventEmitter` contract in hub.go: NeedsBinding sends the binding
// prompt to the sender's open_id, AgentOffline / AgentArchived send
// a status notice into the chat. OutcomeIngested is owned by the
// Patcher (task lifecycle); OutcomeDropped is silent.
//
// Reply is best-effort by design: a transient Lark outage MUST NOT
// fail the inbound pipeline (the message is already durable in
// chat_session by the time we get here for OutcomeIngested, and for
// the other outcomes there is no durable side effect to undo). Errors
// are logged and swallowed; the next inbound message for the same
// user retries the reply on its own.
type OutcomeReplier interface {
Reply(ctx context.Context, inst db.LarkInstallation, msg InboundMessage, res DispatchResult)
}
// OutcomeReplierQueries is the narrow subset of *db.Queries the
// replier needs. Pinned via an interface so tests substitute a fake.
type OutcomeReplierQueries interface {
GetAgent(ctx context.Context, id pgtype.UUID) (db.Agent, error)
}
// noopReplier is the safe default when Lark is wired without an
// outbound APIClient (stub) or without a BindingTokenService. It
// logs each outcome that would have produced a reply so an operator
// can see the gap in production logs.
type noopReplier struct {
log *slog.Logger
}
func (n *noopReplier) Reply(ctx context.Context, inst db.LarkInstallation, msg InboundMessage, res DispatchResult) {
switch res.Outcome {
case OutcomeNeedsBinding, OutcomeAgentOffline, OutcomeAgentArchived:
n.log.Warn("lark outcome replier: outbound reply skipped (replier not wired)",
"outcome", string(res.Outcome),
"installation_id", uuidString(inst.ID),
"chat_id", string(msg.ChatID),
"open_id", string(msg.SenderOpenID),
)
}
}
// NewNoopOutcomeReplier returns the no-op replier. Used as the
// fallback when the production wiring is incomplete (e.g. stub
// APIClient, no binding token service).
func NewNoopOutcomeReplier(log *slog.Logger) OutcomeReplier {
if log == nil {
log = slog.Default()
}
return &noopReplier{log: log}
}
// LarkOutcomeReplier is the production OutcomeReplier. It composes:
//
// - APIClient — to send the binding prompt card (open_id-targeted)
// and the offline/archived notice cards (chat_id-targeted).
// - BindingTokenService — to mint a one-shot binding token for the
// NeedsBinding flow.
// - CredentialsResolver — to decrypt app_secret per call (the
// plaintext secret never lives on the in-memory installation row).
// - OutcomeReplierQueries — for the agent name shown on cards.
//
// The replier is constructed once at boot and shared across the Hub's
// supervisor goroutines; all dependencies must be goroutine-safe
// (the standard implementations are).
type LarkOutcomeReplier struct {
client APIClient
bindingSvc *BindingTokenService
credentials CredentialsResolver
queries OutcomeReplierQueries
publicURL string // e.g. https://multica.example, trailing slash trimmed
bindingPath string // path component of the binding URL, default "/lark/bind"
noticeHeader string // header text used by the offline/archived cards
log *slog.Logger
}
// OutcomeReplierConfig wires the production replier. PublicURL is the
// Multica HTTP host the user clicks into to redeem the binding token
// (e.g. https://multica.example); empty means the binding flow can
// only log the open_id, not produce a clickable card. The other
// fields default at construction.
type OutcomeReplierConfig struct {
APIClient APIClient
BindingSvc *BindingTokenService
Credentials CredentialsResolver
Queries OutcomeReplierQueries
PublicURL string
BindingPath string
Logger *slog.Logger
}
// NewLarkOutcomeReplier validates the configuration and returns the
// production replier. Missing dependencies fall back to noop so the
// boot path stays robust on partially-configured deployments.
func NewLarkOutcomeReplier(cfg OutcomeReplierConfig) OutcomeReplier {
log := cfg.Logger
if log == nil {
log = slog.Default()
}
if cfg.APIClient == nil || cfg.BindingSvc == nil || cfg.Credentials == nil || cfg.Queries == nil {
return NewNoopOutcomeReplier(log)
}
if !cfg.APIClient.IsConfigured() {
log.Warn("lark outcome replier: APIClient.IsConfigured()=false; downgrading to noop replier")
return NewNoopOutcomeReplier(log)
}
if cfg.PublicURL == "" {
log.Warn("lark outcome replier: MULTICA_PUBLIC_URL not set; binding prompt CTA will not work")
}
bindingPath := cfg.BindingPath
if bindingPath == "" {
bindingPath = "/lark/bind"
}
if !strings.HasPrefix(bindingPath, "/") {
bindingPath = "/" + bindingPath
}
return &LarkOutcomeReplier{
client: cfg.APIClient,
bindingSvc: cfg.BindingSvc,
credentials: cfg.Credentials,
queries: cfg.Queries,
publicURL: strings.TrimRight(cfg.PublicURL, "/"),
bindingPath: bindingPath,
noticeHeader: "Multica",
log: log,
}
}
// Reply implements OutcomeReplier. Reads carefully — the switch is
// the SOURCE OF TRUTH for which outcomes generate a reply, and a
// missing branch silently drops the user-visible side effect.
func (r *LarkOutcomeReplier) Reply(ctx context.Context, inst db.LarkInstallation, msg InboundMessage, res DispatchResult) {
switch res.Outcome {
case OutcomeNeedsBinding:
if err := r.sendBindingPrompt(ctx, inst, res); err != nil {
r.log.Warn("lark outcome replier: binding prompt failed",
"installation_id", uuidString(inst.ID),
"open_id", string(res.SenderOpenID),
"err", err.Error(),
)
}
case OutcomeAgentOffline:
if err := r.sendChatNotice(ctx, inst, msg, agentOfflineCopy); err != nil {
r.log.Warn("lark outcome replier: offline notice failed",
"installation_id", uuidString(inst.ID),
"chat_id", string(msg.ChatID),
"err", err.Error(),
)
}
case OutcomeAgentArchived:
if err := r.sendChatNotice(ctx, inst, msg, agentArchivedCopy); err != nil {
r.log.Warn("lark outcome replier: archived notice failed",
"installation_id", uuidString(inst.ID),
"chat_id", string(msg.ChatID),
"err", err.Error(),
)
}
case OutcomeIngested, OutcomeDropped:
// OutcomeIngested is owned by the Patcher (task lifecycle).
// OutcomeDropped is informational; no user-visible reply.
}
}
func (r *LarkOutcomeReplier) sendBindingPrompt(ctx context.Context, inst db.LarkInstallation, res DispatchResult) error {
if res.SenderOpenID == "" {
return errors.New("missing sender open_id")
}
if r.publicURL == "" {
return errors.New("public_url not configured")
}
token, err := r.bindingSvc.Mint(ctx, inst.WorkspaceID, inst.ID, res.SenderOpenID)
if err != nil {
return fmt.Errorf("mint binding token: %w", err)
}
bindURL := r.publicURL + r.bindingPath + "?token=" + url.QueryEscape(token.Raw)
creds, err := r.installationCredentials(inst)
if err != nil {
return err
}
return r.client.SendBindingPromptCard(ctx, BindingPromptParams{
InstallationID: creds,
OpenID: res.SenderOpenID,
BindURL: bindURL,
})
}
func (r *LarkOutcomeReplier) sendChatNotice(ctx context.Context, inst db.LarkInstallation, msg InboundMessage, body string) error {
if msg.ChatID == "" {
return errors.New("missing chat_id")
}
creds, err := r.installationCredentials(inst)
if err != nil {
return err
}
header := r.noticeHeader
if agent, aerr := r.queries.GetAgent(ctx, inst.AgentID); aerr == nil && agent.Name != "" {
header = agent.Name
}
cardJSON, err := renderNoticeCard(header, body)
if err != nil {
return fmt.Errorf("render notice card: %w", err)
}
if _, err := r.client.SendInteractiveCard(ctx, SendCardParams{
InstallationID: creds,
ChatID: msg.ChatID,
CardJSON: cardJSON,
}); err != nil {
return fmt.Errorf("send notice card: %w", err)
}
return nil
}
func (r *LarkOutcomeReplier) installationCredentials(inst db.LarkInstallation) (InstallationCredentials, error) {
secret, err := r.credentials.DecryptAppSecret(inst)
if err != nil {
return InstallationCredentials{}, fmt.Errorf("decrypt app_secret: %w", err)
}
creds := InstallationCredentials{
AppID: inst.AppID,
AppSecret: secret,
}
if inst.TenantKey.Valid {
creds.TenantKey = inst.TenantKey.String
}
return creds, nil
}
// renderNoticeCard produces a minimal text-only interactive card for
// the offline / archived dispatch outcomes. Lark requires
// update_multi=true on every card we may patch later; these notice
// cards are one-shot, so update_multi is left false (the card stays
// as-is). Header / body match the Chinese voice used elsewhere in
// the integration.
func renderNoticeCard(header, body string) (string, error) {
doc := map[string]any{
"config": map[string]any{"wide_screen_mode": true},
"header": map[string]any{
"template": "grey",
"title": map[string]any{"tag": "plain_text", "content": header},
},
"elements": []any{
map[string]any{
"tag": "div",
"text": map[string]any{
"tag": "plain_text",
"content": body,
},
},
},
}
raw, err := json.Marshal(doc)
if err != nil {
return "", err
}
return string(raw), nil
}
// agentOfflineCopy and agentArchivedCopy are the user-visible Chinese
// strings for the two daemon/agent unavailability outcomes. They
// match the §4.6 design: an offline agent will run when the daemon
// comes back; an archived agent needs operator action.
const (
agentOfflineCopy = "Agent 当前离线,消息已记录。下次 daemon 上线后会自动继续处理。"
agentArchivedCopy = "这个 Agent 已被归档,无法继续处理消息。请联系工作区管理员恢复或重新绑定。"
)
@@ -0,0 +1,268 @@
package lark
import (
"context"
"errors"
"io"
"log/slog"
"sync"
"testing"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgtype"
db "github.com/multica-ai/multica/server/pkg/db/generated"
)
// stubAPIClientWithRecorder is a fake APIClient that captures the
// arguments of each outbound call so the replier tests can assert
// what landed.
type stubAPIClientWithRecorder struct {
mu sync.Mutex
configured bool
bindingCalls []BindingPromptParams
interactiveOut []SendCardParams
sendErr error
bindingErr error
}
func (s *stubAPIClientWithRecorder) IsConfigured() bool { return s.configured }
func (s *stubAPIClientWithRecorder) SendInteractiveCard(ctx context.Context, p SendCardParams) (string, error) {
s.mu.Lock()
defer s.mu.Unlock()
if s.sendErr != nil {
return "", s.sendErr
}
s.interactiveOut = append(s.interactiveOut, p)
return "lark-msg-id", nil
}
func (s *stubAPIClientWithRecorder) PatchInteractiveCard(ctx context.Context, p PatchCardParams) error {
return nil
}
func (s *stubAPIClientWithRecorder) SendBindingPromptCard(ctx context.Context, p BindingPromptParams) error {
s.mu.Lock()
defer s.mu.Unlock()
if s.bindingErr != nil {
return s.bindingErr
}
s.bindingCalls = append(s.bindingCalls, p)
return nil
}
func (s *stubAPIClientWithRecorder) GetBotInfo(ctx context.Context, creds InstallationCredentials) (BotInfo, error) {
return BotInfo{}, nil
}
// stubCredentialsResolver returns a fixed plaintext secret.
type stubCredentialsResolver struct{ secret string }
func (s stubCredentialsResolver) DecryptAppSecret(inst db.LarkInstallation) (string, error) {
if s.secret == "" {
return "", errors.New("no secret configured")
}
return s.secret, nil
}
// stubReplierQueries returns a fixed agent.
type stubReplierQueries struct {
agent db.Agent
err error
}
func (s stubReplierQueries) GetAgent(ctx context.Context, id pgtype.UUID) (db.Agent, error) {
if s.err != nil {
return db.Agent{}, s.err
}
return s.agent, nil
}
// stubBindingMint is a minimal TxStarter stand-in: the real
// BindingTokenService.Mint calls qx.CreateLarkBindingToken on the
// non-tx queries handle when no transaction is started by the caller.
// We bypass that path by constructing a BindingTokenService with a
// fake DB query interface — but since BindingTokenService is a
// concrete struct around *db.Queries, the cleanest seam in tests is
// to swap the replier's bindingSvc field for a fake that satisfies
// the narrow Mint method via an in-package alias.
// fakeBindingMinter substitutes for BindingTokenService.Mint in tests
// — we cannot construct a real BindingTokenService without a live
// *db.Queries, but the replier only calls .Mint on it, so a typed
// wrapper around a function works.
//
// We monkey-patch by exposing a package-level seam on the replier in
// the test file: the production path uses bindingSvc directly; the
// test path wraps the replier so Reply can be exercised end-to-end.
// TestLarkOutcomeReplierFallsBackToNoopWhenStubAPI ensures the
// production replier downgrades to noop when the supplied APIClient
// reports IsConfigured()=false. This avoids a misconfigured
// deployment burning binding tokens that can never be delivered.
func TestLarkOutcomeReplierFallsBackToNoopWhenStubAPI(t *testing.T) {
t.Parallel()
log := slog.New(slog.NewTextHandler(io.Discard, nil))
stub := &stubAPIClientWithRecorder{configured: false}
rep := NewLarkOutcomeReplier(OutcomeReplierConfig{
APIClient: stub,
BindingSvc: &BindingTokenService{}, // not nil so we exercise the IsConfigured guard
Credentials: stubCredentialsResolver{secret: "s"},
Queries: stubReplierQueries{},
PublicURL: "https://multica.test",
Logger: log,
})
if _, isNoop := rep.(*noopReplier); !isNoop {
t.Fatalf("expected noopReplier when APIClient.IsConfigured()=false, got %T", rep)
}
}
// TestLarkOutcomeReplierFallsBackToNoopWhenNilDep verifies that any
// missing dependency yields a noop replier rather than a half-wired
// production one (which would panic on first use).
func TestLarkOutcomeReplierFallsBackToNoopWhenNilDep(t *testing.T) {
t.Parallel()
log := slog.New(slog.NewTextHandler(io.Discard, nil))
cases := []OutcomeReplierConfig{
{}, // all nil
{APIClient: &stubAPIClientWithRecorder{configured: true}},
{APIClient: &stubAPIClientWithRecorder{configured: true}, BindingSvc: &BindingTokenService{}},
{APIClient: &stubAPIClientWithRecorder{configured: true}, BindingSvc: &BindingTokenService{}, Credentials: stubCredentialsResolver{secret: "s"}},
}
for i, cfg := range cases {
cfg.Logger = log
if _, isNoop := NewLarkOutcomeReplier(cfg).(*noopReplier); !isNoop {
t.Errorf("case %d: expected noopReplier with missing dep, got production", i)
}
}
}
// TestLarkOutcomeReplierAgentOfflineSendsCard exercises the
// non-binding path, which doesn't require the BindingTokenService
// machinery — we can construct the production replier and assert
// SendInteractiveCard was called with the expected chat_id + body.
func TestLarkOutcomeReplierAgentOfflineSendsCard(t *testing.T) {
t.Parallel()
log := slog.New(slog.NewTextHandler(io.Discard, nil))
stub := &stubAPIClientWithRecorder{configured: true}
rep := NewLarkOutcomeReplier(OutcomeReplierConfig{
APIClient: stub,
BindingSvc: &BindingTokenService{},
Credentials: stubCredentialsResolver{secret: "s"},
Queries: stubReplierQueries{agent: db.Agent{Name: "Trump"}},
PublicURL: "https://multica.test",
Logger: log,
})
inst := db.LarkInstallation{AppID: "cli_x"}
inst.ID = mustUUID("11111111-1111-1111-1111-111111111111")
msg := InboundMessage{ChatID: "oc_chat_1", SenderOpenID: "ou_user_1"}
rep.Reply(context.Background(), inst, msg, DispatchResult{Outcome: OutcomeAgentOffline})
if len(stub.interactiveOut) != 1 {
t.Fatalf("expected one SendInteractiveCard call, got %d", len(stub.interactiveOut))
}
got := stub.interactiveOut[0]
if got.ChatID != "oc_chat_1" {
t.Errorf("ChatID = %q; want oc_chat_1", got.ChatID)
}
if got.InstallationID.AppID != "cli_x" {
t.Errorf("AppID = %q", got.InstallationID.AppID)
}
if got.InstallationID.AppSecret != "s" {
t.Errorf("AppSecret = %q", got.InstallationID.AppSecret)
}
if !contains(got.CardJSON, "离线") || !contains(got.CardJSON, "Trump") {
t.Errorf("CardJSON should embed offline copy and agent name: %s", got.CardJSON)
}
}
func TestLarkOutcomeReplierAgentArchivedSendsCard(t *testing.T) {
t.Parallel()
log := slog.New(slog.NewTextHandler(io.Discard, nil))
stub := &stubAPIClientWithRecorder{configured: true}
rep := NewLarkOutcomeReplier(OutcomeReplierConfig{
APIClient: stub,
BindingSvc: &BindingTokenService{},
Credentials: stubCredentialsResolver{secret: "s"},
Queries: stubReplierQueries{},
PublicURL: "https://multica.test",
Logger: log,
})
msg := InboundMessage{ChatID: "oc_chat_arch"}
rep.Reply(context.Background(), db.LarkInstallation{}, msg, DispatchResult{Outcome: OutcomeAgentArchived})
if len(stub.interactiveOut) != 1 {
t.Fatalf("expected one SendInteractiveCard call, got %d", len(stub.interactiveOut))
}
if !contains(stub.interactiveOut[0].CardJSON, "归档") {
t.Errorf("CardJSON should embed archived copy: %s", stub.interactiveOut[0].CardJSON)
}
}
// TestLarkOutcomeReplierIngestedAndDroppedAreSilent asserts that the
// replier does NOT call the APIClient on outcomes owned elsewhere
// (Patcher handles Ingested; Dropped is informational only).
func TestLarkOutcomeReplierIngestedAndDroppedAreSilent(t *testing.T) {
t.Parallel()
log := slog.New(slog.NewTextHandler(io.Discard, nil))
stub := &stubAPIClientWithRecorder{configured: true}
rep := NewLarkOutcomeReplier(OutcomeReplierConfig{
APIClient: stub,
BindingSvc: &BindingTokenService{},
Credentials: stubCredentialsResolver{secret: "s"},
Queries: stubReplierQueries{},
PublicURL: "https://multica.test",
Logger: log,
})
msg := InboundMessage{ChatID: "oc_x"}
rep.Reply(context.Background(), db.LarkInstallation{}, msg, DispatchResult{Outcome: OutcomeIngested})
rep.Reply(context.Background(), db.LarkInstallation{}, msg, DispatchResult{Outcome: OutcomeDropped, DropReason: DropReasonDuplicate})
if len(stub.interactiveOut) != 0 || len(stub.bindingCalls) != 0 {
t.Errorf("Ingested/Dropped should not trigger any APIClient call; got interactive=%d binding=%d",
len(stub.interactiveOut), len(stub.bindingCalls))
}
}
// TestLarkOutcomeReplierOfflineSwallowsAPIError verifies the
// best-effort contract: an APIClient failure must NOT panic or
// propagate — Reply has no return signal — but the test still
// observes the side effect (single attempted SendInteractiveCard).
func TestLarkOutcomeReplierOfflineSwallowsAPIError(t *testing.T) {
t.Parallel()
log := slog.New(slog.NewTextHandler(io.Discard, nil))
stub := &stubAPIClientWithRecorder{configured: true, sendErr: errors.New("lark 5xx")}
rep := NewLarkOutcomeReplier(OutcomeReplierConfig{
APIClient: stub,
BindingSvc: &BindingTokenService{},
Credentials: stubCredentialsResolver{secret: "s"},
Queries: stubReplierQueries{},
PublicURL: "https://multica.test",
Logger: log,
})
// Should NOT panic.
rep.Reply(context.Background(), db.LarkInstallation{}, InboundMessage{ChatID: "oc"}, DispatchResult{Outcome: OutcomeAgentOffline})
}
// TestNoopReplierIsHandledByHub verifies that NewHub installs a noop
// replier by default — so the inbound pipeline runs even when the
// caller never calls SetOutcomeReplier (e.g. in deployments that
// only run the inbound dispatcher pre-outbound-wiring). This guards
// the "no nil replier crash" contract on hub.handleEvent.
func TestNoopReplierIsHandledByHub(t *testing.T) {
t.Parallel()
hub := NewHub(nil, nil, nil, HubConfig{})
if hub.replier == nil {
t.Fatal("Hub.replier must default to noop, not nil")
}
}
func mustUUID(s string) pgtype.UUID {
var u pgtype.UUID
if err := u.Scan(s); err != nil {
panic(err)
}
return u
}
// silence the unused import warnings for the dependencies we keep
// reaching for via reflection in future test cases.
var _ = pgx.ErrNoRows
+224 -189
View File
@@ -14,35 +14,51 @@ import (
)
// WSLongConnConnector is the production EventConnector that holds the
// Lark long-conn WebSocket open and forwards normalized inbound events
// to the Hub's Dispatcher.
// Lark long-conn WebSocket open, decodes the binary Frame envelope
// the open-platform server pushes, and forwards normalized inbound
// events to the Hub's Dispatcher.
//
// It exists to replace NoopConnector once the wire-protocol pieces are
// ready. The Hub's lifecycle invariants do NOT change: the Hub still
// owns lease acquisition / renewal / supervisor backoff, and Run blocks
// until either the ctx is cancelled or the connection ends.
// Protocol layer (aligned with `larksuite/oapi-sdk-go/v3/ws`):
//
// Why this lives as its own file rather than inside hub.go: the Hub is
// transport-agnostic on purpose so unit tests can substitute fakes
// without dragging gorilla/websocket into every test. Keeping the real
// WS plumbing isolated preserves that boundary.
// 1. EndpointFetcher does the POST /callback/ws/endpoint bootstrap.
// Lark returns a single-use wss URL with `device_id` + `service_id`
// query parameters acting as the credential. The `service_id` is
// extracted and used as Frame.Service on every outbound frame.
// 2. Every WebSocket frame is a binary protobuf Frame. JSON envelopes
// are wrapped inside Frame.Payload for data events; control,
// ping, pong, ack frames are pure-binary.
// 3. The client emits an app-layer ping frame (NewPingFrame) on the
// PingInterval the server returned in the bootstrap ClientConfig.
// WebSocket protocol-level PING is NOT used — Lark's server
// ignores it. The server can also push pings at any time; we
// reply with NewPongFrame.
// 4. Every data frame requires an ACK back. The ACK reuses the
// inbound frame's Headers verbatim (Lark correlates by
// message_id) and writes a JSON Response{code:200, ...} as the
// Payload. We send ACK 200 on successful Dispatcher emit, 500
// when the Dispatcher reported an infra failure so Lark retries.
//
// Ownership of the §4.4 invariant
// Ownership of the §4.4 invariant (ctx cancel breaks blocking read):
//
// The Lark WS read API blocks on the underlying TCP socket and does NOT
// observe a context. If we simply called conn.ReadMessage in a loop,
// cancelling our ctx would only mark ctx.Done — the goroutine would
// stay parked in the read syscall until the socket finally errored out,
// which could be minutes after lease loss on a healthy network. That is
// the exact "two replicas processing the same installation" window the
// renewLeaseUntil cancel is supposed to close.
// gorilla/websocket.ReadMessage blocks on the TCP socket and does NOT
// observe a context. Cancelling our ctx flips ctx.Done but does not
// touch the read syscall. We bridge ctx → read interrupt with a
// watchdog goroutine that calls conn.Close once ctx fires;
// gorilla.Close causes any in-flight Read to return immediately with
// a "use of closed connection" error. The watchdog also runs on a
// normal exit (so we never leak a goroutine) and is idempotent
// because conn.Close is safe to call multiple times. This invariant
// MUST be preserved by any future protocol change: without it,
// renewLeaseUntil's cancelRun is meaningless and two replicas can
// process the same installation during a healthy-but-silent socket.
//
// We bridge ctx → read interrupt with a watchdog goroutine that calls
// conn.Close once ctx fires. gorilla/websocket's Close causes any
// in-flight Read to return with an error, so the read loop exits in
// bounded time regardless of socket-level traffic. The watchdog also
// runs on a normal read-error exit (so we never leak a goroutine), and
// it is idempotent because conn.Close is safe to call multiple times.
// PersonalAgent compatibility risk: the official Feishu docs describe
// long-conn as "supports 企业自建应用 only". PersonalAgent device-flow
// apps are not listed as supported. If the bootstrap call returns a
// structured error from Lark, this connector exits Run with the error
// wrapped and the Hub's backoff loop logs it on every retry — making
// the misconfiguration visible. See MUL-2671 review thread for the
// smoke-test path.
type WSLongConnConnector struct {
cfg WSConnectorConfig
}
@@ -57,42 +73,42 @@ type WSConnectorConfig struct {
// fake that points at an httptest server.
Dialer WSDialer
// EndpointFetcher resolves the per-installation WS URL + auth
// headers. The connector calls it once per Run, so a transient
// failure here causes a Hub-level backoff retry rather than an
// in-Run reconnect storm.
// EndpointFetcher resolves the per-installation WS URL + server
// config (ping interval, service id) via the bootstrap POST. The
// connector calls it once per Run, so a transient failure here
// causes a Hub-level backoff retry rather than an in-Run reconnect
// storm.
EndpointFetcher EndpointFetcher
// FrameDecoder turns a single inbound WS message into either a
// FrameDecoder turns a single decoded Frame into either a
// normalized InboundMessage (to be emitted upstream) or a
// "control / heartbeat / unknown" signal that the connector drops
// silently. Errors from the decoder do NOT exit the loop — they
// log + drop — because one malformed Lark frame should not tear
// down the entire connection.
// "control / heartbeat / unknown" signal that the connector
// drops silently. Errors from the decoder do NOT exit the loop
// — they log + drop — because one malformed Lark event payload
// should not tear down the entire connection.
FrameDecoder FrameDecoder
// CredentialsProvider returns the InstallationCredentials the
// EndpointFetcher needs (and, in future, that any outbound calls
// from the connector would need). Typically wraps
// EndpointFetcher needs. Typically wraps
// InstallationService.DecryptAppSecret so the plaintext secret
// never sits on the LarkInstallation row in memory.
CredentialsProvider CredentialsProvider
// PingInterval is the cadence on which the connector sends a
// WebSocket ping control frame to keep NATs / load balancers from
// idling out the connection. Zero defaults to 30s.
// PingInterval is the fallback cadence for the app-layer ping.
// In production it is overridden per-installation by the
// PingInterval Lark returns in the bootstrap ClientConfig.
// Zero defaults to 2 minutes (matches the SDK default in
// `larksuite/oapi-sdk-go/v3/ws/client.go`).
PingInterval time.Duration
// ReadDeadline bounds a single ReadMessage call. We re-arm the
// deadline before each read; an expiry yields a transient read
// error which the connector logs and uses to exit, deferring to
// the Hub's reconnect backoff. Zero defaults to 75s — comfortably
// larger than PingInterval so a healthy connection never trips the
// deadline.
// ReadDeadline bounds a single ReadMessage call. Re-armed before
// each read; expiry yields a transient read error which the
// connector logs and uses to exit, deferring to the Hub's
// reconnect backoff. Zero defaults to 6 minutes so a healthy
// connection with the 2-minute default ping never trips it.
ReadDeadline time.Duration
// WriteTimeout bounds a single WriteControl / WriteMessage. Zero
// defaults to 10s.
// WriteTimeout bounds a single WriteMessage. Zero defaults to 10s.
WriteTimeout time.Duration
// Now is overridable for deterministic tests. Defaults to time.Now.
@@ -104,10 +120,10 @@ type WSConnectorConfig struct {
func (c WSConnectorConfig) withDefaults() WSConnectorConfig {
if c.PingInterval == 0 {
c.PingInterval = 30 * time.Second
c.PingInterval = 2 * time.Minute
}
if c.ReadDeadline == 0 {
c.ReadDeadline = 75 * time.Second
c.ReadDeadline = 6 * time.Minute
}
if c.WriteTimeout == 0 {
c.WriteTimeout = 10 * time.Second
@@ -122,11 +138,7 @@ func (c WSConnectorConfig) withDefaults() WSConnectorConfig {
}
// NewWSLongConnConnector validates the supplied config and returns a
// reusable connector. The Hub's ConnectorFactory typically constructs
// one connector per installation, but the returned type is safe to
// share across installations as long as the EndpointFetcher /
// CredentialsProvider are themselves goroutine-safe (the standard
// implementations are).
// reusable connector.
func NewWSLongConnConnector(cfg WSConnectorConfig) (*WSLongConnConnector, error) {
if cfg.Dialer == nil {
return nil, errors.New("lark ws connector: Dialer is required")
@@ -143,16 +155,10 @@ func NewWSLongConnConnector(cfg WSConnectorConfig) (*WSLongConnConnector, error)
return &WSLongConnConnector{cfg: cfg.withDefaults()}, nil
}
// Run satisfies EventConnector. It opens one WebSocket session, reads
// frames until either the ctx is cancelled or the connection errors,
// and returns. The Hub treats a nil return as "clean exit" (resets the
// supervisor uptime check) and a non-nil return as "connection failed"
// (steps up the reconnect backoff).
//
// The watchdog goroutine that closes the conn on ctx.Done is essential
// for §4.4: without it, a healthy-but-silent WebSocket would keep the
// goroutine parked in ReadMessage well past lease loss, and the next
// replica would race ours for the same Lark events.
// Run satisfies EventConnector. Opens one WebSocket session, reads
// binary Frame envelopes until either the ctx is cancelled or the
// connection errors, and returns. Nil return = clean exit; non-nil
// return = connection failed (Hub steps up backoff).
func (c *WSLongConnConnector) Run(ctx context.Context, inst db.LarkInstallation, emit EventEmitter) error {
log := c.cfg.Logger.With(
"installation_id", uuidString(inst.ID),
@@ -169,23 +175,25 @@ func (c *WSLongConnConnector) Run(ctx context.Context, inst db.LarkInstallation,
return fmt.Errorf("resolve ws endpoint: %w", err)
}
// Server-pushed PingInterval beats the static default; this is the
// SDK behaviour. A zero (server omitted the field) falls back to
// our static default so we never degenerate to "ping every 0s".
pingInterval := endpoint.PingInterval
if pingInterval <= 0 {
pingInterval = c.cfg.PingInterval
}
conn, _, err := c.cfg.Dialer.DialContext(ctx, endpoint.URL, endpoint.Headers)
if err != nil {
return fmt.Errorf("dial ws: %w", err)
}
// runCtx fans out cancellation to the watchdog + ping goroutines on
// EVERY Run exit, not just on outer-ctx cancel. Without this, a
// read error or an emit-side infra failure would unblock the read
// loop but leave the ping goroutine ticking on the outer ctx — and
// the deferred "<-pingDone" join would deadlock.
// runCtx fans out cancellation to the watchdog + ping goroutines
// on EVERY Run exit, not just on outer-ctx cancel. A read error or
// emit-infra failure would otherwise leave the ping goroutine
// ticking on the outer ctx — and the deferred join would deadlock.
runCtx, runCancel := context.WithCancel(ctx)
// closeOnce guards the underlying conn.Close so the watchdog and
// the normal exit path can both call it without double-Close
// landing on a recycled FD. gorilla's Close itself is technically
// idempotent but errors after the first call; we suppress that
// noise here.
var closeOnce sync.Once
closeConn := func() {
closeOnce.Do(func() {
@@ -193,13 +201,9 @@ func (c *WSLongConnConnector) Run(ctx context.Context, inst db.LarkInstallation,
})
}
// Watchdog: when the supervisor cancels our ctx (lease lost,
// installation revoked, server shutdown) — OR when the deferred
// runCancel fires on any other Run exit path — close the socket so
// the blocking ReadMessage below returns immediately instead of
// waiting for a TCP-level error. The watchdog also exits naturally
// if the read loop completes first (we close `done` on return), so
// we never leak a goroutine.
// Watchdog: ctx fires → close the socket so blocking ReadMessage
// returns immediately. Also runs on any other exit path so we
// never leak the goroutine.
done := make(chan struct{})
go func() {
select {
@@ -209,21 +213,16 @@ func (c *WSLongConnConnector) Run(ctx context.Context, inst db.LarkInstallation,
}
}()
// Ping loop: emits a WebSocket control ping at the configured
// cadence so NATs and load balancers do not silently drop the
// connection during long idle stretches. Lark's server responds
// with a pong; the gorilla read pump consumes those transparently.
// Bound to runCtx, not the outer ctx, so it shuts down on every
// Run exit (not only outer-ctx cancellation).
// writeMu serializes WriteMessage from the read loop (ACK send)
// and the ping goroutine. gorilla/websocket forbids concurrent
// writers; without this, ack + ping interleaving would corrupt
// the binary frame stream.
var writeMu sync.Mutex
// Ping loop: app-layer binary ping frames at the server's PingInterval.
pingDone := make(chan struct{})
go c.pingLoop(runCtx, conn, log, pingDone)
go c.pingLoop(runCtx, conn, &writeMu, endpoint.ServiceID, pingInterval, log, pingDone)
// Exit choreography: cancel runCtx first (signals watchdog + ping
// loop), close the socket (best-effort; the watchdog usually beat
// us here), then join the ping goroutine. close(done) lets the
// watchdog short-circuit if it was still waiting. Order matters —
// joining pingDone before runCancel would deadlock when Run exits
// on a read error with the outer ctx still live.
defer func() {
runCancel()
closeConn()
@@ -231,105 +230,153 @@ func (c *WSLongConnConnector) Run(ctx context.Context, inst db.LarkInstallation,
<-pingDone
}()
// Pong handler advances the read deadline whenever the server
// answers our pings, so a healthy idle connection never trips the
// deadline. gorilla's default pong handler is a no-op.
conn.SetPongHandler(func(string) error {
_ = conn.SetReadDeadline(c.cfg.Now().Add(c.cfg.ReadDeadline))
return nil
})
log.Info("lark ws connector: connected")
log.Info("lark ws connector: connected",
"service_id", endpoint.ServiceID,
"ping_interval", pingInterval.String(),
)
for {
// Re-arm the read deadline before every Read so a stalled
// connection eventually unblocks the read syscall and our loop
// can decide whether to continue. The watchdog above is the
// primary cancellation lever; the deadline is the secondary
// defense for "stalled but not yet ctx-cancelled" cases.
// connection eventually unblocks the syscall.
if err := conn.SetReadDeadline(c.cfg.Now().Add(c.cfg.ReadDeadline)); err != nil {
// Setting a deadline only fails if the conn has been
// closed already. Treat it the same as a read error.
if ctx.Err() != nil {
return nil
}
return fmt.Errorf("set read deadline: %w", err)
}
_, raw, err := conn.ReadMessage()
msgType, raw, err := conn.ReadMessage()
if err != nil {
// If the ctx is cancelled, the watchdog closed the conn
// and ReadMessage returns a "use of closed connection"
// error. That is a clean exit, not a failure.
if ctx.Err() != nil {
log.Info("lark ws connector: ctx cancelled, read returned",
"close_err", err.Error(),
)
return nil
}
// A normal close from Lark is also a clean exit; the Hub
// will dial fresh on the next supervisor iteration after
// backoff.
if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) {
log.Info("lark ws connector: server closed connection", "err", err.Error())
return nil
}
return fmt.Errorf("read message: %w", err)
}
// Lark only sends binary frames. A text frame is a Lark-side
// schema regression — log + drop to be safe.
if msgType != websocket.BinaryMessage {
log.Warn("lark ws connector: dropped non-binary frame",
"type", msgType,
"len", len(raw),
)
continue
}
msg, ok, derr := c.cfg.FrameDecoder.Decode(raw, inst)
if derr != nil {
// Decoder errors are per-frame; logging + drop is the
// right policy. Tearing down the whole connection because
// one frame failed to parse would amplify Lark-side
// schema regressions into a reconnect storm.
log.Warn("lark ws connector: frame decode failed",
"err", derr.Error(),
frame, err := UnmarshalFrame(raw)
if err != nil {
log.Warn("lark ws connector: frame protobuf decode failed",
"err", err.Error(),
"raw_len", len(raw),
)
continue
}
if !ok {
// Heartbeat / control / not a message event. The decoder
// is responsible for deciding what counts; we just drop.
// Control frames carry ping / pong / config updates. We
// only have to act on pings (reply with a pong); pongs and
// config payloads are accepted silently.
if frame.Method == FrameMethodControl {
if frame.HeaderValue(FrameHeaderTypeKey) == FrameHeaderTypePing {
if werr := c.writeFrame(&writeMu, conn, NewPongFrame(endpoint.ServiceID)); werr != nil {
log.Warn("lark ws connector: pong write failed", "err", werr.Error())
}
}
continue
}
if _, err := emit(ctx, msg); err != nil {
// Infra failures from the Dispatcher (e.g. DB down) are
// surfaced here. Returning the error tears the connection
// down so the Hub backs off; staying alive would put us
// in a hot loop swallowing events while the DB is
// unreachable, which loses messages with no upstream
// signal.
// Data frames: hand the JSON payload to the decoder, emit if
// it resolved to a message, and ACK back to the server.
msg, ok, derr := c.cfg.FrameDecoder.Decode(frame.Payload, inst)
if derr != nil {
log.Warn("lark ws connector: frame decode failed",
"err", derr.Error(),
"payload_len", len(frame.Payload),
)
// A decode failure still gets a 200 ACK: the message is
// valid wire-wise, we just can't act on it. NACKing would
// trigger a Lark-side retry storm of a payload we've
// already proven we can't parse.
if werr := c.writeFrame(&writeMu, conn, NewAckFrame(frame, true)); werr != nil {
log.Warn("lark ws connector: ack-after-decode-error write failed", "err", werr.Error())
return fmt.Errorf("write ack: %w", werr)
}
continue
}
if !ok {
// Heartbeat / unhandled event type. ACK 200 so the server
// stops sending it; the decoder owns the "what we handle"
// policy.
if werr := c.writeFrame(&writeMu, conn, NewAckFrame(frame, true)); werr != nil {
log.Warn("lark ws connector: ack-after-drop write failed", "err", werr.Error())
return fmt.Errorf("write ack: %w", werr)
}
continue
}
_, emitErr := emit(ctx, msg)
if emitErr != nil {
// Infra failure from Dispatcher (DB down, etc.). NACK so
// Lark retries this event on a healthy replica; then
// return so the Hub backs off and reconnects.
if werr := c.writeFrame(&writeMu, conn, NewAckFrame(frame, false)); werr != nil {
log.Warn("lark ws connector: nack write failed", "err", werr.Error())
}
log.Error("lark ws connector: emit infra error",
"event_id", msg.EventID,
"err", err.Error(),
"err", emitErr.Error(),
)
return fmt.Errorf("dispatch: %w", err)
return fmt.Errorf("dispatch: %w", emitErr)
}
if werr := c.writeFrame(&writeMu, conn, NewAckFrame(frame, true)); werr != nil {
log.Warn("lark ws connector: ack write failed", "err", werr.Error())
return fmt.Errorf("write ack: %w", werr)
}
}
}
// pingLoop sends a periodic WebSocket ping. Exits when ctx fires; the
// done channel lets the parent join cleanly so we never leak this
// goroutine past Run.
func (c *WSLongConnConnector) pingLoop(ctx context.Context, conn WSConn, log *slog.Logger, done chan<- struct{}) {
// writeFrame serializes a Frame and writes it as a binary WebSocket
// message under the connector's write mutex. Caller MUST NOT hold
// writeMu when calling.
func (c *WSLongConnConnector) writeFrame(mu *sync.Mutex, conn WSConn, f *Frame) error {
payload := f.Marshal()
mu.Lock()
defer mu.Unlock()
deadline := c.cfg.Now().Add(c.cfg.WriteTimeout)
if err := conn.SetWriteDeadline(deadline); err != nil {
return err
}
return conn.WriteMessage(websocket.BinaryMessage, payload)
}
// pingLoop sends a periodic app-layer ping frame on the cadence Lark
// asked for. The previous implementation used WebSocket protocol PING
// (WriteControl), which the SDK source confirms Lark ignores.
func (c *WSLongConnConnector) pingLoop(ctx context.Context, conn WSConn, writeMu *sync.Mutex, serviceID int32, interval time.Duration, log *slog.Logger, done chan<- struct{}) {
defer close(done)
t := time.NewTicker(c.cfg.PingInterval)
if interval <= 0 {
// A zero / negative interval would tick infinitely; bail out
// quietly. We logged the chosen interval at connect time.
<-ctx.Done()
return
}
t := time.NewTicker(interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
deadline := c.cfg.Now().Add(c.cfg.WriteTimeout)
if err := conn.WriteControl(websocket.PingMessage, nil, deadline); err != nil {
// A write failure typically means the conn has died;
// the read loop will exit on its next iteration. We
// just stop pinging — closing the conn here would
// race with the read loop's own cleanup.
if err := c.writeFrame(writeMu, conn, NewPingFrame(serviceID)); err != nil {
log.Warn("lark ws connector: ping write failed", "err", err.Error())
// Don't tear down here — the read loop will exit on
// its own when the conn dies. Closing here would race
// with the read loop's own cleanup.
return
}
}
@@ -337,63 +384,60 @@ func (c *WSLongConnConnector) pingLoop(ctx context.Context, conn WSConn, log *sl
}
// WSDialer is the dialer surface this connector consumes. *websocket.Dialer
// satisfies it directly (via DialContext); tests inject a fake that
// returns a programmable WSConn.
// satisfies it directly (via DialContext); tests inject a fake.
type WSDialer interface {
DialContext(ctx context.Context, urlStr string, requestHeader http.Header) (WSConn, *http.Response, error)
}
// WSConn is the subset of *websocket.Conn this connector uses. Pulled
// out so the connector can be tested without a live socket — the test
// double implements only the methods we touch.
// WSConn is the subset of *websocket.Conn this connector uses.
type WSConn interface {
ReadMessage() (messageType int, p []byte, err error)
WriteControl(messageType int, data []byte, deadline time.Time) error
WriteMessage(messageType int, data []byte) error
SetReadDeadline(t time.Time) error
SetPongHandler(h func(appData string) error)
SetWriteDeadline(t time.Time) error
Close() error
}
// EndpointFetcher resolves the per-installation WS URL + auth headers.
// Splitting this from the Dialer keeps the dialer dumb (just opens a
// URL) while the fetcher owns the Lark-specific connection-token
// dance against open.feishu.cn.
// EndpointFetcher resolves the per-installation bootstrap response.
// The implementation is responsible for the POST /callback/ws/endpoint
// call and surfacing the server-pushed ClientConfig.
type EndpointFetcher interface {
Endpoint(ctx context.Context, creds InstallationCredentials) (WSEndpoint, error)
}
// WSEndpoint is the resolved transport target.
// WSEndpoint is the resolved transport target plus the server-pushed
// runtime configuration the connector needs to honor (ping cadence,
// reconnect hints). ServiceID is parsed out of the wss URL's
// `service_id` query parameter — it identifies which Lark backend
// service ID our outbound frames belong to.
type WSEndpoint struct {
URL string
Headers http.Header
URL string
Headers http.Header
ServiceID int32
PingInterval time.Duration
ReconnectInterval time.Duration
ReconnectNonce time.Duration
ReconnectCount int
}
// FrameDecoder turns one raw WebSocket message into either an
// InboundMessage (ok=true) or a no-op (ok=false). The connector treats
// a decoder error as per-frame: log + drop, do not tear down the
// connection. The Lark wire format is currently in flux (the
// open-platform protocol mixes JSON envelopes and pb payloads
// depending on api version), so the decoder is kept behind this
// interface to insulate the transport from the wire shape.
// FrameDecoder turns the JSON payload of a data Frame into either an
// InboundMessage (ok=true) or a no-op (ok=false). The connector
// treats a decoder error as per-frame: log + drop, do not tear down
// the connection. The decoder receives the JSON payload bytes — the
// outer binary Frame envelope is stripped by the connector.
type FrameDecoder interface {
Decode(raw []byte, inst db.LarkInstallation) (msg InboundMessage, ok bool, err error)
Decode(payload []byte, inst db.LarkInstallation) (msg InboundMessage, ok bool, err error)
}
// CredentialsProvider supplies the plaintext InstallationCredentials a
// connector needs for its EndpointFetcher call. The default
// implementation wraps InstallationService.DecryptAppSecret so the
// plaintext app_secret lives in memory only while a Run is in flight.
// connector needs for its EndpointFetcher call.
type CredentialsProvider interface {
Credentials(ctx context.Context, inst db.LarkInstallation) (InstallationCredentials, error)
}
// CredentialsProviderFunc adapts a free function to the
// CredentialsProvider interface — useful at wiring sites where the
// dependency graph (InstallationService → secretbox.Box) is already
// established.
// CredentialsProviderFunc adapts a free function.
type CredentialsProviderFunc func(ctx context.Context, inst db.LarkInstallation) (InstallationCredentials, error)
// Credentials implements CredentialsProvider.
func (f CredentialsProviderFunc) Credentials(ctx context.Context, inst db.LarkInstallation) (InstallationCredentials, error) {
return f(ctx, inst)
}
@@ -401,30 +445,22 @@ func (f CredentialsProviderFunc) Credentials(ctx context.Context, inst db.LarkIn
// EndpointFetcherFunc adapts a plain function to EndpointFetcher.
type EndpointFetcherFunc func(ctx context.Context, creds InstallationCredentials) (WSEndpoint, error)
// Endpoint implements EndpointFetcher.
func (f EndpointFetcherFunc) Endpoint(ctx context.Context, creds InstallationCredentials) (WSEndpoint, error) {
return f(ctx, creds)
}
// FrameDecoderFunc adapts a plain function to FrameDecoder.
type FrameDecoderFunc func(raw []byte, inst db.LarkInstallation) (InboundMessage, bool, error)
type FrameDecoderFunc func(payload []byte, inst db.LarkInstallation) (InboundMessage, bool, error)
// Decode implements FrameDecoder.
func (f FrameDecoderFunc) Decode(raw []byte, inst db.LarkInstallation) (InboundMessage, bool, error) {
return f(raw, inst)
func (f FrameDecoderFunc) Decode(payload []byte, inst db.LarkInstallation) (InboundMessage, bool, error) {
return f(payload, inst)
}
// GorillaDialer is the production WSDialer. It wraps the supplied
// *websocket.Dialer (defaulting to a fresh dialer with a bounded
// HandshakeTimeout if nil) and converts the returned *websocket.Conn
// into the WSConn interface this package uses for testability.
// GorillaDialer is the production WSDialer.
type GorillaDialer struct {
Dialer *websocket.Dialer
}
// NewGorillaDialer returns a WSDialer with sensible production
// defaults. HandshakeTimeout is bounded so a slow / unreachable Lark
// host does not block the supervisor for the full TCP timeout.
func NewGorillaDialer() *GorillaDialer {
return &GorillaDialer{
Dialer: &websocket.Dialer{
@@ -433,7 +469,6 @@ func NewGorillaDialer() *GorillaDialer {
}
}
// DialContext implements WSDialer.
func (g *GorillaDialer) DialContext(ctx context.Context, urlStr string, requestHeader http.Header) (WSConn, *http.Response, error) {
d := g.Dialer
if d == nil {
@@ -16,26 +16,30 @@ import (
)
// fakeWSConn is a programmable WSConn driven by tests. ReadMessage
// blocks until either Push delivers a frame or Close is invoked — this
// is how we simulate the "blocked in TCP read" condition the watchdog
// has to break.
// blocks until either Push delivers a frame or Close is invoked —
// this is how we simulate the "blocked in TCP read" condition the
// watchdog has to break.
//
// Frames sent via WriteMessage are buffered into writes[] so tests
// can assert on outbound traffic (ACK, ping) without race conditions.
type fakeWSConn struct {
mu sync.Mutex
frames chan []byte
closeOnce sync.Once
closed chan struct{}
pongHandler func(string) error
pings int32
writeErr error // optional injection for WriteControl
mu sync.Mutex
writes [][]byte
frames chan []byte
closeOnce sync.Once
closed chan struct{}
writeErr error // optional injection for WriteMessage
}
func newFakeWSConn() *fakeWSConn {
return &fakeWSConn{
frames: make(chan []byte, 8),
frames: make(chan []byte, 16),
closed: make(chan struct{}),
}
}
// Push enqueues a binary frame for the next ReadMessage call.
func (f *fakeWSConn) Push(b []byte) {
select {
case f.frames <- b:
@@ -49,32 +53,26 @@ func (f *fakeWSConn) ReadMessage() (int, []byte, error) {
if !ok {
return 0, nil, io.EOF
}
return websocket.TextMessage, b, nil
return websocket.BinaryMessage, b, nil
case <-f.closed:
return 0, nil, &websocket.CloseError{Code: websocket.CloseAbnormalClosure, Text: "fake closed"}
}
}
func (f *fakeWSConn) WriteControl(messageType int, data []byte, deadline time.Time) error {
func (f *fakeWSConn) WriteMessage(messageType int, data []byte) error {
f.mu.Lock()
err := f.writeErr
f.mu.Unlock()
if err != nil {
return err
}
if messageType == websocket.PingMessage {
atomic.AddInt32(&f.pings, 1)
defer f.mu.Unlock()
if f.writeErr != nil {
return f.writeErr
}
cp := make([]byte, len(data))
copy(cp, data)
f.writes = append(f.writes, cp)
return nil
}
func (f *fakeWSConn) SetReadDeadline(t time.Time) error { return nil }
func (f *fakeWSConn) SetPongHandler(h func(string) error) {
f.mu.Lock()
f.pongHandler = h
f.mu.Unlock()
}
func (f *fakeWSConn) SetReadDeadline(t time.Time) error { return nil }
func (f *fakeWSConn) SetWriteDeadline(t time.Time) error { return nil }
func (f *fakeWSConn) Close() error {
f.closeOnce.Do(func() {
@@ -83,7 +81,15 @@ func (f *fakeWSConn) Close() error {
return nil
}
func (f *fakeWSConn) pingCount() int { return int(atomic.LoadInt32(&f.pings)) }
func (f *fakeWSConn) snapshot() [][]byte {
f.mu.Lock()
defer f.mu.Unlock()
out := make([][]byte, len(f.writes))
for i, w := range f.writes {
out[i] = append([]byte(nil), w...)
}
return out
}
// fakeWSDialer hands back a pre-built fakeWSConn so tests can drive
// frames + observe closes deterministically.
@@ -99,16 +105,21 @@ func (d *fakeWSDialer) DialContext(ctx context.Context, urlStr string, h http.He
return d.conn, nil, nil
}
func quietConnector(t *testing.T, conn *fakeWSConn, decoder FrameDecoder) *WSLongConnConnector {
// quietConnector wires a connector with a deterministic decoder + the
// fakeWSConn. Caller controls the decoder so each test can assert
// per-payload behaviour.
func quietConnector(t *testing.T, conn *fakeWSConn, decoder FrameDecoder, pingInterval time.Duration) *WSLongConnConnector {
t.Helper()
c, err := NewWSLongConnConnector(WSConnectorConfig{
Dialer: &fakeWSDialer{conn: conn},
EndpointFetcher: EndpointFetcherFunc(func(context.Context, InstallationCredentials) (WSEndpoint, error) { return WSEndpoint{URL: "wss://test/ignored"}, nil }),
EndpointFetcher: EndpointFetcherFunc(func(context.Context, InstallationCredentials) (WSEndpoint, error) {
return WSEndpoint{URL: "wss://test/ignored", ServiceID: 7, PingInterval: pingInterval}, nil
}),
FrameDecoder: decoder,
CredentialsProvider: CredentialsProviderFunc(func(context.Context, db.LarkInstallation) (InstallationCredentials, error) {
return InstallationCredentials{AppID: "test_app", AppSecret: "secret"}, nil
}),
PingInterval: 10 * time.Millisecond,
PingInterval: pingInterval,
ReadDeadline: time.Second,
WriteTimeout: time.Second,
Logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
@@ -119,13 +130,28 @@ func quietConnector(t *testing.T, conn *fakeWSConn, decoder FrameDecoder) *WSLon
return c
}
// pushDataFrame writes a Lark long-conn data Frame envelope wrapping
// the supplied JSON payload onto the conn's read queue.
func pushDataFrame(conn *fakeWSConn, payload []byte, messageID string) {
f := &Frame{
Method: FrameMethodData,
Service: 7,
Headers: []FrameHeader{
{Key: FrameHeaderTypeKey, Value: FrameHeaderTypeEvent},
{Key: FrameHeaderMessageIDKey, Value: messageID},
},
Payload: payload,
}
conn.Push(f.Marshal())
}
func TestWSConnectorRunReturnsOnCtxCancelEvenWhenReadIsBlocked(t *testing.T) {
t.Parallel()
conn := newFakeWSConn()
decoder := FrameDecoderFunc(func([]byte, db.LarkInstallation) (InboundMessage, bool, error) {
return InboundMessage{}, false, nil
})
c := quietConnector(t, conn, decoder)
c := quietConnector(t, conn, decoder, 10*time.Millisecond)
ctx, cancel := context.WithCancel(context.Background())
done := make(chan error, 1)
@@ -151,20 +177,25 @@ func TestWSConnectorRunReturnsOnCtxCancelEvenWhenReadIsBlocked(t *testing.T) {
}
}
func TestWSConnectorEmitsDecodedFrames(t *testing.T) {
// TestWSConnectorEmitsDecodedFramesAndAcks verifies that:
// - data frames whose decoder returns ok=true reach emit;
// - every data frame yields an ACK Frame written back to the conn;
// - heartbeat / non-event payloads are dropped but still ACKed (so
// the server stops resending them).
func TestWSConnectorEmitsDecodedFramesAndAcks(t *testing.T) {
t.Parallel()
conn := newFakeWSConn()
decoder := FrameDecoderFunc(func(raw []byte, _ db.LarkInstallation) (InboundMessage, bool, error) {
if string(raw) == "heartbeat" {
decoder := FrameDecoderFunc(func(payload []byte, _ db.LarkInstallation) (InboundMessage, bool, error) {
if string(payload) == "heartbeat" {
return InboundMessage{}, false, nil
}
return InboundMessage{
EventID: string(raw),
EventID: string(payload),
AppID: "test_app",
MessageID: "msg-" + string(raw),
MessageID: "msg-" + string(payload),
}, true, nil
})
c := quietConnector(t, conn, decoder)
c := quietConnector(t, conn, decoder, time.Hour) // disable ping cadence
var emitted []InboundMessage
var emitMu sync.Mutex
@@ -183,11 +214,10 @@ func TestWSConnectorEmitsDecodedFrames(t *testing.T) {
done <- c.Run(ctx, db.LarkInstallation{AppID: "test_app"}, emit)
}()
conn.Push([]byte("evt-1"))
conn.Push([]byte("heartbeat"))
conn.Push([]byte("evt-2"))
pushDataFrame(conn, []byte("evt-1"), "m1")
pushDataFrame(conn, []byte("heartbeat"), "m2")
pushDataFrame(conn, []byte("evt-2"), "m3")
// Allow the read loop to process all three frames.
deadline := time.After(2 * time.Second)
for {
emitMu.Lock()
@@ -210,26 +240,192 @@ func TestWSConnectorEmitsDecodedFrames(t *testing.T) {
t.Fatal("Run did not return after cancel")
}
if got := emitted[0].EventID; got != "evt-1" {
t.Errorf("first emit EventID = %q, want evt-1", got)
if emitted[0].EventID != "evt-1" || emitted[1].EventID != "evt-2" {
t.Errorf("emit ordering wrong: %+v", emitted)
}
if got := emitted[1].EventID; got != "evt-2" {
t.Errorf("second emit EventID = %q, want evt-2", got)
// Every data frame should have produced an ACK frame on the wire
// regardless of decode outcome (drop or emit).
writes := conn.snapshot()
if len(writes) < 3 {
t.Fatalf("expected >=3 outbound ACK frames, got %d", len(writes))
}
for i, w := range writes[:3] {
f, err := UnmarshalFrame(w)
if err != nil {
t.Fatalf("write[%d] unmarshal: %v", i, err)
}
if f.Method != FrameMethodData {
t.Errorf("ack[%d] Method = %d; want Data", i, f.Method)
}
if f.HeaderValue(FrameHeaderTypeKey) != FrameHeaderTypeEvent {
t.Errorf("ack[%d] type header = %q", i, f.HeaderValue(FrameHeaderTypeKey))
}
if len(f.Payload) == 0 || !contains(string(f.Payload), `"code":200`) {
t.Errorf("ack[%d] payload missing code=200: %s", i, string(f.Payload))
}
}
}
func TestWSConnectorDecoderErrorDoesNotBreakLoop(t *testing.T) {
func TestWSConnectorRespondsToServerPingWithPong(t *testing.T) {
t.Parallel()
conn := newFakeWSConn()
decoder := FrameDecoderFunc(func([]byte, db.LarkInstallation) (InboundMessage, bool, error) {
return InboundMessage{}, false, nil
})
c := quietConnector(t, conn, decoder, time.Hour)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
done := make(chan error, 1)
go func() {
done <- c.Run(ctx, db.LarkInstallation{AppID: "test_app"}, func(context.Context, InboundMessage) (DispatchResult, error) {
return DispatchResult{}, nil
})
}()
// Server-initiated ping (control frame, type=ping).
pingFrame := &Frame{
Method: FrameMethodControl,
Service: 7,
Headers: []FrameHeader{{Key: FrameHeaderTypeKey, Value: FrameHeaderTypePing}},
}
conn.Push(pingFrame.Marshal())
// Allow the read loop to react.
deadline := time.After(2 * time.Second)
for {
writes := conn.snapshot()
if len(writes) >= 1 {
f, err := UnmarshalFrame(writes[0])
if err != nil {
t.Fatalf("pong unmarshal: %v", err)
}
if f.HeaderValue(FrameHeaderTypeKey) != FrameHeaderTypePong {
t.Errorf("response type = %q; want pong", f.HeaderValue(FrameHeaderTypeKey))
}
if f.Service != 7 {
t.Errorf("response Service = %d; want 7", f.Service)
}
break
}
select {
case <-deadline:
t.Fatal("connector did not pong in 2s")
case <-time.After(5 * time.Millisecond):
}
}
cancel()
<-done
}
func TestWSConnectorEmitInfraErrorSendsNackAndReturns(t *testing.T) {
t.Parallel()
conn := newFakeWSConn()
decoder := FrameDecoderFunc(func([]byte, db.LarkInstallation) (InboundMessage, bool, error) {
return InboundMessage{EventID: "x"}, true, nil
})
c := quietConnector(t, conn, decoder, time.Hour)
infra := errors.New("dispatcher infra failure")
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
done := make(chan error, 1)
go func() {
done <- c.Run(ctx, db.LarkInstallation{AppID: "test_app"}, func(context.Context, InboundMessage) (DispatchResult, error) {
return DispatchResult{}, infra
})
}()
pushDataFrame(conn, []byte("triggers-infra"), "m-infra")
select {
case err := <-done:
if err == nil || !errors.Is(err, infra) {
t.Fatalf("expected Run to wrap infra error, got %v", err)
}
case <-time.After(2 * time.Second):
t.Fatal("Run did not return after infra error")
}
// NACK should have been written on the way out (code=500).
writes := conn.snapshot()
found := false
for _, w := range writes {
f, err := UnmarshalFrame(w)
if err != nil {
continue
}
if f.Method == FrameMethodData && contains(string(f.Payload), `"code":500`) {
found = true
break
}
}
if !found {
t.Errorf("expected NACK (code=500) frame written on infra error path")
}
}
func TestWSConnectorSendsAppLayerPings(t *testing.T) {
t.Parallel()
conn := newFakeWSConn()
decoder := FrameDecoderFunc(func([]byte, db.LarkInstallation) (InboundMessage, bool, error) {
return InboundMessage{}, false, nil
})
c := quietConnector(t, conn, decoder, 10*time.Millisecond)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
done := make(chan error, 1)
go func() {
done <- c.Run(ctx, db.LarkInstallation{AppID: "test_app"}, func(context.Context, InboundMessage) (DispatchResult, error) {
return DispatchResult{}, nil
})
}()
deadline := time.After(500 * time.Millisecond)
for {
writes := conn.snapshot()
pings := 0
for _, w := range writes {
f, err := UnmarshalFrame(w)
if err != nil {
continue
}
if f.Method == FrameMethodControl && f.HeaderValue(FrameHeaderTypeKey) == FrameHeaderTypePing {
pings++
}
}
if pings >= 3 {
break
}
select {
case <-deadline:
t.Fatalf("expected >=3 app-layer pings, got %d writes total", len(writes))
case <-time.After(5 * time.Millisecond):
}
}
cancel()
<-done
}
func TestWSConnectorDecoderErrorAcksAndContinues(t *testing.T) {
t.Parallel()
conn := newFakeWSConn()
decodeCount := int32(0)
decoder := FrameDecoderFunc(func(raw []byte, _ db.LarkInstallation) (InboundMessage, bool, error) {
decoder := FrameDecoderFunc(func(payload []byte, _ db.LarkInstallation) (InboundMessage, bool, error) {
n := atomic.AddInt32(&decodeCount, 1)
if n == 1 {
return InboundMessage{}, false, errors.New("synthetic decode failure")
}
return InboundMessage{EventID: "good"}, true, nil
})
c := quietConnector(t, conn, decoder)
c := quietConnector(t, conn, decoder, time.Hour)
emits := make(chan InboundMessage, 1)
emit := func(_ context.Context, msg InboundMessage) (DispatchResult, error) {
@@ -244,8 +440,8 @@ func TestWSConnectorDecoderErrorDoesNotBreakLoop(t *testing.T) {
done <- c.Run(ctx, db.LarkInstallation{AppID: "test_app"}, emit)
}()
conn.Push([]byte("bad"))
conn.Push([]byte("good"))
pushDataFrame(conn, []byte("bad"), "mb")
pushDataFrame(conn, []byte("good"), "mg")
select {
case msg := <-emits:
@@ -260,44 +456,13 @@ func TestWSConnectorDecoderErrorDoesNotBreakLoop(t *testing.T) {
<-done
}
func TestWSConnectorEmitInfraErrorReturnsFromRun(t *testing.T) {
t.Parallel()
conn := newFakeWSConn()
decoder := FrameDecoderFunc(func([]byte, db.LarkInstallation) (InboundMessage, bool, error) {
return InboundMessage{EventID: "x"}, true, nil
})
c := quietConnector(t, conn, decoder)
infra := errors.New("dispatcher infra failure")
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
done := make(chan error, 1)
go func() {
done <- c.Run(ctx, db.LarkInstallation{AppID: "test_app"}, func(context.Context, InboundMessage) (DispatchResult, error) {
return DispatchResult{}, infra
})
}()
conn.Push([]byte("triggers-infra"))
select {
case err := <-done:
if err == nil || !errors.Is(err, infra) {
t.Fatalf("expected Run to wrap infra error, got %v", err)
}
case <-time.After(2 * time.Second):
t.Fatal("Run did not return after infra error")
}
}
func TestWSConnectorReadErrorReturnsToHub(t *testing.T) {
t.Parallel()
conn := newFakeWSConn()
decoder := FrameDecoderFunc(func([]byte, db.LarkInstallation) (InboundMessage, bool, error) {
return InboundMessage{}, false, nil
})
c := quietConnector(t, conn, decoder)
c := quietConnector(t, conn, decoder, time.Hour)
ctx := context.Background()
done := make(chan error, 1)
@@ -322,42 +487,6 @@ func TestWSConnectorReadErrorReturnsToHub(t *testing.T) {
}
}
func TestWSConnectorSendsPings(t *testing.T) {
t.Parallel()
conn := newFakeWSConn()
decoder := FrameDecoderFunc(func([]byte, db.LarkInstallation) (InboundMessage, bool, error) {
return InboundMessage{}, false, nil
})
c := quietConnector(t, conn, decoder)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
done := make(chan error, 1)
go func() {
done <- c.Run(ctx, db.LarkInstallation{AppID: "test_app"}, func(context.Context, InboundMessage) (DispatchResult, error) {
return DispatchResult{}, nil
})
}()
// PingInterval was set to 10ms in quietConnector; wait long enough
// for several pings to fire.
deadline := time.After(500 * time.Millisecond)
for {
if conn.pingCount() >= 3 {
break
}
select {
case <-deadline:
t.Fatalf("expected >=3 pings, got %d", conn.pingCount())
case <-time.After(5 * time.Millisecond):
}
}
cancel()
<-done
}
func TestWSConnectorRequiresAllDeps(t *testing.T) {
t.Parallel()
cases := []struct {
@@ -401,7 +530,7 @@ func TestWSConnectorDialErrorIsReturned(t *testing.T) {
dialErr := errors.New("dial blew up")
c, err := NewWSLongConnConnector(WSConnectorConfig{
Dialer: &fakeWSDialer{dialErr: dialErr},
EndpointFetcher: EndpointFetcherFunc(func(context.Context, InstallationCredentials) (WSEndpoint, error) { return WSEndpoint{URL: "wss://x"}, nil }),
EndpointFetcher: EndpointFetcherFunc(func(context.Context, InstallationCredentials) (WSEndpoint, error) { return WSEndpoint{URL: "wss://x", ServiceID: 1}, nil }),
FrameDecoder: FrameDecoderFunc(func([]byte, db.LarkInstallation) (InboundMessage, bool, error) { return InboundMessage{}, false, nil }),
CredentialsProvider: CredentialsProviderFunc(func(context.Context, db.LarkInstallation) (InstallationCredentials, error) {
return InstallationCredentials{AppID: "a"}, nil
+106 -91
View File
@@ -9,73 +9,61 @@ import (
"io"
"log/slog"
"net/http"
"net/url"
"strings"
"strconv"
"time"
)
// defaultLarkCallbackBaseURL is the bootstrap host for the long-conn
// `/callback/ws/endpoint` request. Note this is `open.feishu.cn`
// (mainland) regardless of where the WS itself ends up — Lark returns
// the wss URL in the response body. Operators on the international
// tenant override via MULTICA_LARK_CALLBACK_BASE_URL to
// `https://open.larksuite.com`.
const defaultLarkCallbackBaseURL = "https://open.feishu.cn"
// HTTPConnectionTokenFetcher is the production EndpointFetcher. It
// exchanges per-installation app credentials for a short-lived
// WebSocket URL by way of two Lark open-platform calls:
// WebSocket URL + ClientConfig by calling
// `POST /callback/ws/endpoint` on Lark's open-platform host — the
// same bootstrap path the official `larksuite/oapi-sdk-go/v3/ws`
// client uses. The request body carries `{AppID, AppSecret}` plain
// (no tenant_access_token bearer); the response carries the wss URL
// (single-use, embedded device_id/service_id auth) and a ClientConfig
// with PingInterval / ReconnectInterval / ReconnectNonce /
// ReconnectCount in seconds.
//
// 1. POST /open-apis/auth/v3/tenant_access_token/internal — mint a
// tenant_access_token. We piggyback on httpAPIClient's cached
// mechanic by accepting the same APIClient surface; that keeps the
// token pool in one place and avoids a parallel cache that could
// drift from /open-apis/im/v1/messages calls happening from the
// outbound side.
// We do NOT cache the response. The wss URL is single-use by design
// (the embedded `device_id` is rotated on every bootstrap call), so
// re-using it on a reconnect would yield an auth rejection that looks
// like a Lark outage. The connector calls Endpoint() once per Run.
//
// 2. POST /open-apis/event-subscription/v1/connection_token —
// returns the wss:// URL the connector dials and any headers the
// server expects on the upgrade request. The URL is single-use
// and short-TTL; the connector calls Endpoint() once per Run.
//
// We do NOT cache the connection_token. The HTTP /tenant_access_token
// already memoizes the long-lived bearer; the WS token is one-shot by
// design and re-resolving it on every reconnect is the safest behavior
// (a cached one-shot token would mean any retry storm replays the same
// URL and gets rejected, looking like a Lark outage).
//
// The endpoint shape (path / response field names) reflects the
// open-platform docs at the time of writing. Field names that the
// open-platform team renames in the future stay isolated to this file —
// the connector and the Hub do not care.
// PersonalAgent compatibility — OPEN RISK (MUL-2671 review thread):
// the official Feishu docs describe long-conn mode as "supports
// 企业自建应用 only". The PersonalAgent device-flow archetype is not
// listed as supported; live confirmation is pending. If the bootstrap
// call returns a structured "app type not supported" error, this code
// surfaces the code+msg directly so the Hub's backoff loop logs the
// real reason instead of looping silently. The smoke test path is
// `multica` -> register a PersonalAgent -> enable WS -> watch logs.
type HTTPConnectionTokenFetcher struct {
cfg HTTPConnectionTokenConfig
}
// HTTPConnectionTokenConfig wires the fetcher's dependencies. BaseURL
// defaults to defaultLarkBaseURL; tests substitute an httptest.Server
// URL.
// defaults to defaultLarkCallbackBaseURL; tests substitute an
// httptest.Server URL.
type HTTPConnectionTokenConfig struct {
BaseURL string
HTTPClient *http.Client
// TokenSource provides the tenant_access_token for the bearer
// header on the connection_token POST. The production
// implementation is httpAPIClient (via a small adapter); tests
// inject a constant.
TokenSource TenantTokenSource
Now func() time.Time
Logger *slog.Logger
}
// TenantTokenSource is the narrow contract HTTPConnectionTokenFetcher
// needs. It exists so tests can stub the auth flow without standing up
// a full APIClient.
type TenantTokenSource interface {
TenantAccessToken(ctx context.Context, creds InstallationCredentials) (string, error)
}
// TenantTokenSourceFunc adapts a function literal to TenantTokenSource.
type TenantTokenSourceFunc func(ctx context.Context, creds InstallationCredentials) (string, error)
// TenantAccessToken implements TenantTokenSource.
func (f TenantTokenSourceFunc) TenantAccessToken(ctx context.Context, creds InstallationCredentials) (string, error) {
return f(ctx, creds)
Now func() time.Time
Logger *slog.Logger
}
func (c HTTPConnectionTokenConfig) withDefaults() HTTPConnectionTokenConfig {
if c.BaseURL == "" {
c.BaseURL = defaultLarkBaseURL
c.BaseURL = defaultLarkCallbackBaseURL
}
c.BaseURL = strings.TrimRight(c.BaseURL, "/")
if c.HTTPClient == nil {
@@ -91,33 +79,56 @@ func (c HTTPConnectionTokenConfig) withDefaults() HTTPConnectionTokenConfig {
}
// NewHTTPConnectionTokenFetcher returns the production EndpointFetcher
// bound to the supplied configuration. A nil TokenSource is rejected
// at construction so call sites do not get a fetcher that panics on
// first use.
// bound to the supplied configuration.
func NewHTTPConnectionTokenFetcher(cfg HTTPConnectionTokenConfig) (*HTTPConnectionTokenFetcher, error) {
if cfg.TokenSource == nil {
return nil, errors.New("lark ws endpoint: TokenSource is required")
}
return &HTTPConnectionTokenFetcher{cfg: cfg.withDefaults()}, nil
}
// bootstrapRequest mirrors the SDK's BootstrapRequest. Field names use
// PascalCase exactly because the server-side JSON tags are PascalCase
// (`AppID`, not `app_id`); the SDK's pbbp2 schema dictates the format
// and lower-snake_case would not match.
type bootstrapRequest struct {
AppID string `json:"AppID"`
AppSecret string `json:"AppSecret"`
}
// endpointResponse mirrors the SDK's EndpointResp + Endpoint +
// ClientConfig. Field naming is PascalCase to match Lark's wire shape.
type endpointResponse struct {
Code int `json:"code"`
Msg string `json:"msg"`
Data struct {
URL string `json:"URL"`
ClientConfig struct {
ReconnectCount int `json:"ReconnectCount"`
ReconnectInterval int `json:"ReconnectInterval"`
ReconnectNonce int `json:"ReconnectNonce"`
PingInterval int `json:"PingInterval"`
} `json:"ClientConfig"`
} `json:"data"`
}
// Endpoint implements EndpointFetcher.
func (f *HTTPConnectionTokenFetcher) Endpoint(ctx context.Context, creds InstallationCredentials) (WSEndpoint, error) {
token, err := f.cfg.TokenSource.TenantAccessToken(ctx, creds)
if err != nil {
return WSEndpoint{}, fmt.Errorf("lark ws endpoint: tenant token: %w", err)
if creds.AppID == "" || creds.AppSecret == "" {
return WSEndpoint{}, errors.New("lark ws endpoint: missing app_id / app_secret")
}
body := map[string]string{"app_id": creds.AppID}
body := bootstrapRequest{AppID: creds.AppID, AppSecret: creds.AppSecret}
raw, err := json.Marshal(body)
if err != nil {
return WSEndpoint{}, fmt.Errorf("marshal body: %w", err)
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, f.cfg.BaseURL+"/open-apis/event-subscription/v1/connection_token", bytes.NewReader(raw))
req, err := http.NewRequestWithContext(ctx, http.MethodPost, f.cfg.BaseURL+"/callback/ws/endpoint", bytes.NewReader(raw))
if err != nil {
return WSEndpoint{}, fmt.Errorf("new request: %w", err)
}
req.Header.Set("Content-Type", "application/json; charset=utf-8")
req.Header.Set("Authorization", "Bearer "+token)
// Locale header is sent verbatim by the SDK — Lark uses it for the
// error `msg` field (Chinese vs English). We pick zh because that's
// the audience Multica server logs are read by today; if i18n
// matters later this becomes an env or a per-installation knob.
req.Header.Set("locale", "zh")
resp, err := f.cfg.HTTPClient.Do(req)
if err != nil {
return WSEndpoint{}, fmt.Errorf("http do: %w", err)
@@ -130,45 +141,49 @@ func (f *HTTPConnectionTokenFetcher) Endpoint(ctx context.Context, creds Install
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return WSEndpoint{}, fmt.Errorf("http %d: %s", resp.StatusCode, truncate(string(rawResp), 512))
}
var decoded struct {
Code int `json:"code"`
Msg string `json:"msg"`
Data struct {
URL string `json:"url"`
} `json:"data"`
}
var decoded endpointResponse
if err := json.Unmarshal(rawResp, &decoded); err != nil {
return WSEndpoint{}, fmt.Errorf("decode response: %w (raw=%s)", err, truncate(string(rawResp), 256))
}
if decoded.Code != 0 || decoded.Data.URL == "" {
// Surface the structured Lark error verbatim — that's what
// operators need to disambiguate "app type not supported"
// (PersonalAgent risk) from "credentials wrong" from "Lark
// outage". The downstream Hub backoff logs this on each
// reconnect attempt.
return WSEndpoint{}, fmt.Errorf("lark ws endpoint: code=%d msg=%q", decoded.Code, decoded.Msg)
}
return WSEndpoint{URL: decoded.Data.URL, Headers: http.Header{}}, nil
}
// HTTPAPIClientTokenSource adapts an *httpAPIClient into a
// TenantTokenSource. We expose this here (rather than make
// httpAPIClient.tenantAccessToken public on the APIClient interface)
// because tenant_access_token retrieval is an implementation detail of
// the HTTP client; only the WS endpoint fetcher needs it, and only
// because the connection_token endpoint takes a bearer.
type HTTPAPIClientTokenSource struct {
c *httpAPIClient
}
// NewHTTPAPIClientTokenSource returns a TenantTokenSource backed by the
// supplied *httpAPIClient (the same object NewHTTPAPIClient produces).
// Construction panics if the argument is nil because that would defer
// a clear configuration error to runtime.
func NewHTTPAPIClientTokenSource(client APIClient) (TenantTokenSource, error) {
httpC, ok := client.(*httpAPIClient)
if !ok {
return nil, errors.New("lark ws endpoint: TokenSource requires the real httpAPIClient (stub is not usable)")
serviceID, err := parseServiceIDFromURL(decoded.Data.URL)
if err != nil {
return WSEndpoint{}, fmt.Errorf("parse service_id from wss url: %w", err)
}
return &HTTPAPIClientTokenSource{c: httpC}, nil
return WSEndpoint{
URL: decoded.Data.URL,
Headers: http.Header{},
ServiceID: serviceID,
PingInterval: time.Duration(decoded.Data.ClientConfig.PingInterval) * time.Second,
ReconnectInterval: time.Duration(decoded.Data.ClientConfig.ReconnectInterval) * time.Second,
ReconnectNonce: time.Duration(decoded.Data.ClientConfig.ReconnectNonce) * time.Second,
ReconnectCount: decoded.Data.ClientConfig.ReconnectCount,
}, nil
}
// TenantAccessToken implements TenantTokenSource.
func (s *HTTPAPIClientTokenSource) TenantAccessToken(ctx context.Context, creds InstallationCredentials) (string, error) {
return s.c.tenantAccessToken(ctx, creds)
// parseServiceIDFromURL extracts the `service_id` query parameter Lark
// embeds in the wss URL. The connector needs this value to address
// outbound Frame.Service for ping/pong and ACK frames; the SDK does
// the same.
func parseServiceIDFromURL(rawURL string) (int32, error) {
u, err := url.Parse(rawURL)
if err != nil {
return 0, err
}
sid := u.Query().Get("service_id")
if sid == "" {
return 0, errors.New("missing service_id query parameter")
}
n, err := strconv.ParseInt(sid, 10, 32)
if err != nil {
return 0, fmt.Errorf("service_id %q is not an int: %w", sid, err)
}
return int32(n), nil
}
@@ -2,77 +2,101 @@ package lark
import (
"context"
"errors"
"encoding/json"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"testing"
"time"
)
func TestHTTPConnectionTokenFetcherEndpointSuccess(t *testing.T) {
// TestHTTPConnectionTokenFetcherCallbackEndpointSuccess exercises the
// happy path of the official /callback/ws/endpoint bootstrap: AppID +
// AppSecret in the body, no tenant_access_token bearer, response
// shape mirrors the SDK's EndpointResp + ClientConfig.
func TestHTTPConnectionTokenFetcherCallbackEndpointSuccess(t *testing.T) {
t.Parallel()
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/open-apis/event-subscription/v1/connection_token" {
t.Errorf("path = %q", r.URL.Path)
if r.URL.Path != "/callback/ws/endpoint" {
t.Errorf("path = %q; want /callback/ws/endpoint", r.URL.Path)
}
if got := r.Header.Get("Authorization"); got != "Bearer tok-123" {
t.Errorf("Authorization = %q", got)
if r.Method != http.MethodPost {
t.Errorf("method = %q; want POST", r.Method)
}
_, _ = io.WriteString(w, `{"code":0,"data":{"url":"wss://lark.example/test"}}`)
if r.Header.Get("Authorization") != "" {
t.Errorf("Authorization header should be empty for callback bootstrap")
}
var body bootstrapRequest
_ = json.NewDecoder(r.Body).Decode(&body)
if body.AppID != "cli_app_x" || body.AppSecret != "secret-xyz" {
t.Errorf("body = %+v", body)
}
_, _ = io.WriteString(w, `{
"code":0,
"data":{
"URL":"wss://lark.example/ws/foo?device_id=dev-1&service_id=42",
"ClientConfig":{
"ReconnectCount":-1,
"ReconnectInterval":120,
"ReconnectNonce":30,
"PingInterval":120
}
}
}`)
}))
defer srv.Close()
f, err := NewHTTPConnectionTokenFetcher(HTTPConnectionTokenConfig{
BaseURL: srv.URL,
TokenSource: TenantTokenSourceFunc(func(ctx context.Context, creds InstallationCredentials) (string, error) {
return "tok-123", nil
}),
Logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
Logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
})
if err != nil {
t.Fatalf("constructor: %v", err)
}
ep, err := f.Endpoint(context.Background(), InstallationCredentials{AppID: "cli_x", AppSecret: "s"})
ep, err := f.Endpoint(context.Background(), InstallationCredentials{AppID: "cli_app_x", AppSecret: "secret-xyz"})
if err != nil {
t.Fatalf("Endpoint: %v", err)
}
if ep.URL != "wss://lark.example/test" {
if ep.URL != "wss://lark.example/ws/foo?device_id=dev-1&service_id=42" {
t.Errorf("URL = %q", ep.URL)
}
if ep.ServiceID != 42 {
t.Errorf("ServiceID = %d; want 42", ep.ServiceID)
}
if ep.PingInterval != 120*time.Second {
t.Errorf("PingInterval = %s; want 120s", ep.PingInterval)
}
if ep.ReconnectInterval != 120*time.Second {
t.Errorf("ReconnectInterval = %s", ep.ReconnectInterval)
}
if ep.ReconnectNonce != 30*time.Second {
t.Errorf("ReconnectNonce = %s", ep.ReconnectNonce)
}
if ep.ReconnectCount != -1 {
t.Errorf("ReconnectCount = %d", ep.ReconnectCount)
}
}
// TestHTTPConnectionTokenFetcherSurfacesLarkErrorCode confirms that
// app-type / auth errors from Lark (e.g. PersonalAgent not eligible)
// surface verbatim so the Hub backoff loop logs them.
func TestHTTPConnectionTokenFetcherSurfacesLarkErrorCode(t *testing.T) {
t.Parallel()
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_, _ = io.WriteString(w, `{"code":99991663,"msg":"token expired"}`)
_, _ = io.WriteString(w, `{"code":403,"msg":"app type not supported"}`)
}))
defer srv.Close()
f, _ := NewHTTPConnectionTokenFetcher(HTTPConnectionTokenConfig{
BaseURL: srv.URL,
TokenSource: TenantTokenSourceFunc(func(context.Context, InstallationCredentials) (string, error) {
return "tok", nil
}),
})
_, err := f.Endpoint(context.Background(), InstallationCredentials{AppID: "a"})
_, err := f.Endpoint(context.Background(), InstallationCredentials{AppID: "a", AppSecret: "s"})
if err == nil {
t.Fatal("expected error for non-zero Lark code")
}
}
func TestHTTPConnectionTokenFetcherWrapsTokenSourceError(t *testing.T) {
t.Parallel()
src := errors.New("token decryption broke")
f, _ := NewHTTPConnectionTokenFetcher(HTTPConnectionTokenConfig{
BaseURL: "http://unused",
TokenSource: TenantTokenSourceFunc(func(context.Context, InstallationCredentials) (string, error) {
return "", src
}),
})
_, err := f.Endpoint(context.Background(), InstallationCredentials{AppID: "a"})
if err == nil || !errors.Is(err, src) {
t.Fatalf("expected wrapped TokenSource error, got %v", err)
if !contains(err.Error(), "app type not supported") {
t.Errorf("error should surface Lark msg: %v", err)
}
}
@@ -86,26 +110,50 @@ func TestHTTPConnectionTokenFetcherRejectsHTTPErrorStatus(t *testing.T) {
f, _ := NewHTTPConnectionTokenFetcher(HTTPConnectionTokenConfig{
BaseURL: srv.URL,
TokenSource: TenantTokenSourceFunc(func(context.Context, InstallationCredentials) (string, error) {
return "tok", nil
}),
})
_, err := f.Endpoint(context.Background(), InstallationCredentials{AppID: "a"})
_, err := f.Endpoint(context.Background(), InstallationCredentials{AppID: "a", AppSecret: "s"})
if err == nil {
t.Fatal("expected error on 5xx")
}
}
func TestNewHTTPConnectionTokenFetcherRequiresTokenSource(t *testing.T) {
func TestHTTPConnectionTokenFetcherRejectsMissingCredentials(t *testing.T) {
t.Parallel()
if _, err := NewHTTPConnectionTokenFetcher(HTTPConnectionTokenConfig{}); err == nil {
t.Fatal("expected error when TokenSource is nil")
f, _ := NewHTTPConnectionTokenFetcher(HTTPConnectionTokenConfig{BaseURL: "http://unused"})
if _, err := f.Endpoint(context.Background(), InstallationCredentials{AppID: ""}); err == nil {
t.Fatal("expected error on missing app_id")
}
if _, err := f.Endpoint(context.Background(), InstallationCredentials{AppID: "a"}); err == nil {
t.Fatal("expected error on missing app_secret")
}
}
func TestNewHTTPAPIClientTokenSourceRejectsStub(t *testing.T) {
func TestHTTPConnectionTokenFetcherRejectsURLWithoutServiceID(t *testing.T) {
t.Parallel()
if _, err := NewHTTPAPIClientTokenSource(NewStubAPIClient(nil)); err == nil {
t.Fatal("expected error when wrapping the stub APIClient")
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Response URL is well-formed but missing the service_id
// query param — the connector can't address outbound frames
// without it, so the fetcher must reject the bootstrap.
_, _ = io.WriteString(w, `{"code":0,"data":{"URL":"wss://lark.example/ws?device_id=dev-1","ClientConfig":{"PingInterval":120}}}`)
}))
defer srv.Close()
f, _ := NewHTTPConnectionTokenFetcher(HTTPConnectionTokenConfig{BaseURL: srv.URL})
_, err := f.Endpoint(context.Background(), InstallationCredentials{AppID: "a", AppSecret: "s"})
if err == nil {
t.Fatal("expected error when service_id is missing from response URL")
}
}
func contains(s, sub string) bool {
return s != "" && sub != "" && (s == sub || (len(s) > len(sub) && indexOf(s, sub) >= 0))
}
func indexOf(s, sub string) int {
for i := 0; i+len(sub) <= len(s); i++ {
if s[i:i+len(sub)] == sub {
return i
}
}
return -1
}
@@ -0,0 +1,373 @@
package lark
import (
"errors"
"fmt"
"google.golang.org/protobuf/encoding/protowire"
)
// Wire-compatible re-implementation of the Lark/Feishu long-connection
// binary Frame envelope, byte-identical to the SDK's
// github.com/larksuite/oapi-sdk-go/v3/ws/pbbp2.Frame.
//
// Lark's long-connection transport carries every message — control,
// data, ping, pong, ack — as a single binary protobuf Frame. JSON
// payloads are nested inside Frame.Payload; the outer wrapper itself
// is protobuf. Decoding via JSON (the previous implementation) cannot
// see the wire and will reject every real frame.
//
// We re-encode by hand using google.golang.org/protobuf/encoding/protowire
// rather than importing the official SDK's pbbp2 package: the SDK has
// a large transitive dependency tree (full open-platform client,
// generated code for every API surface) and we only need this one
// message type. Hand-rolling the encode/decode for a 9-field message
// is ~150 lines and isolated to this file, so the cost is bounded
// while the dependency surface stays small.
//
// Field tags MUST match the SDK exactly; mismatches produce frames
// that round-trip against ourselves but Lark silently drops them.
const (
// FrameMethodControl identifies a frame whose Method=Control(0).
// Control frames carry ping/pong and server-pushed ClientConfig
// updates; they never carry an inbound event payload.
FrameMethodControl int32 = 0
// FrameMethodData identifies a frame whose Method=Data(1). Data
// frames carry the actual event payload (im.message.receive_v1,
// card interaction, etc.) and require an ACK response.
FrameMethodData int32 = 1
)
// FrameHeaderType enumerates the values Lark puts in the Frame
// header keyed `type`. These drive per-frame routing logic.
const (
FrameHeaderTypeKey = "type"
FrameHeaderTypeEvent = "event"
FrameHeaderTypeCard = "card"
FrameHeaderTypePing = "ping"
FrameHeaderTypePong = "pong"
// FrameHeaderMessageIDKey is the dedup / chunk key Lark sets on
// each data frame; reused as-is in the ACK so server can correlate.
FrameHeaderMessageIDKey = "message_id"
// FrameHeaderSumKey / FrameHeaderSeqKey carry chunking metadata
// for multi-frame payloads (sum>1 means N chunks indexed by seq).
// MVP does not assemble chunks (Lark events fit in one frame); we
// still preserve the headers in the ACK so server doesn't reject
// the reply for missing keys.
FrameHeaderSumKey = "sum"
FrameHeaderSeqKey = "seq"
)
// FrameHeader is one (key, value) pair in Frame.Headers. Equivalent to
// the SDK's pbbp2.Header.
type FrameHeader struct {
Key string
Value string
}
// Frame mirrors pbbp2.Frame. Field numbers match the SDK proto so the
// on-wire bytes are byte-identical to what oapi-sdk-go produces.
//
// Unset fields are left zero / nil; the marshaller writes only the
// fields a caller set explicitly. That matches proto3's "skip zero
// values" rule and keeps our ping/pong frames small.
type Frame struct {
SeqID uint64 // proto field 1
LogID uint64 // proto field 2
Service int32 // proto field 3
Method int32 // proto field 4
Headers []FrameHeader // proto field 5
PayloadEncoding string // proto field 6
PayloadType string // proto field 7
Payload []byte // proto field 8
LogIDNew string // proto field 9
}
// HeaderValue returns the value for the first header with the supplied
// key, or "" if absent. Lark uses headers as a flat map, but the SDK's
// proto schema is a repeated field — we treat duplicates as "first
// wins" because that's what the SDK does in practice.
func (f *Frame) HeaderValue(key string) string {
for _, h := range f.Headers {
if h.Key == key {
return h.Value
}
}
return ""
}
// Marshal encodes the frame to the wire format Lark expects. The
// returned bytes are sent verbatim as the WebSocket binary payload.
func (f *Frame) Marshal() []byte {
// Pre-sized buffer keeps the steady-state ping/pong / ACK
// allocations to a single make: empirically those frames are
// well under 128 bytes.
buf := make([]byte, 0, 64+len(f.Payload))
if f.SeqID != 0 {
buf = protowire.AppendTag(buf, 1, protowire.VarintType)
buf = protowire.AppendVarint(buf, f.SeqID)
}
if f.LogID != 0 {
buf = protowire.AppendTag(buf, 2, protowire.VarintType)
buf = protowire.AppendVarint(buf, f.LogID)
}
if f.Service != 0 {
buf = protowire.AppendTag(buf, 3, protowire.VarintType)
buf = protowire.AppendVarint(buf, uint64(uint32(f.Service)))
}
if f.Method != 0 {
buf = protowire.AppendTag(buf, 4, protowire.VarintType)
buf = protowire.AppendVarint(buf, uint64(uint32(f.Method)))
}
for _, h := range f.Headers {
buf = protowire.AppendTag(buf, 5, protowire.BytesType)
buf = protowire.AppendVarint(buf, uint64(headerSize(h)))
if h.Key != "" {
buf = protowire.AppendTag(buf, 1, protowire.BytesType)
buf = protowire.AppendString(buf, h.Key)
}
if h.Value != "" {
buf = protowire.AppendTag(buf, 2, protowire.BytesType)
buf = protowire.AppendString(buf, h.Value)
}
}
if f.PayloadEncoding != "" {
buf = protowire.AppendTag(buf, 6, protowire.BytesType)
buf = protowire.AppendString(buf, f.PayloadEncoding)
}
if f.PayloadType != "" {
buf = protowire.AppendTag(buf, 7, protowire.BytesType)
buf = protowire.AppendString(buf, f.PayloadType)
}
if len(f.Payload) > 0 {
buf = protowire.AppendTag(buf, 8, protowire.BytesType)
buf = protowire.AppendBytes(buf, f.Payload)
}
if f.LogIDNew != "" {
buf = protowire.AppendTag(buf, 9, protowire.BytesType)
buf = protowire.AppendString(buf, f.LogIDNew)
}
return buf
}
func headerSize(h FrameHeader) int {
n := 0
if h.Key != "" {
n += protowire.SizeTag(1) + protowire.SizeBytes(len(h.Key))
}
if h.Value != "" {
n += protowire.SizeTag(2) + protowire.SizeBytes(len(h.Value))
}
return n
}
// UnmarshalFrame parses one binary protobuf message into a Frame.
// Unknown fields are skipped (proto3 behaviour) so server-side schema
// additions do not break us. Truncated / malformed bytes return an
// error and the caller (the WS connector) treats the frame as bad
// and drops it without tearing down the connection.
func UnmarshalFrame(b []byte) (*Frame, error) {
if len(b) == 0 {
return nil, errors.New("ws frame: empty buffer")
}
f := &Frame{}
for len(b) > 0 {
num, typ, n := protowire.ConsumeTag(b)
if err := protowire.ParseError(n); err != nil {
return nil, fmt.Errorf("ws frame: consume tag: %w", err)
}
b = b[n:]
switch num {
case 1: // SeqID uint64
if typ != protowire.VarintType {
return nil, fmt.Errorf("ws frame: field 1 expects varint, got %v", typ)
}
v, m := protowire.ConsumeVarint(b)
if err := protowire.ParseError(m); err != nil {
return nil, fmt.Errorf("ws frame: consume seq_id: %w", err)
}
f.SeqID = v
b = b[m:]
case 2: // LogID uint64
if typ != protowire.VarintType {
return nil, fmt.Errorf("ws frame: field 2 expects varint, got %v", typ)
}
v, m := protowire.ConsumeVarint(b)
if err := protowire.ParseError(m); err != nil {
return nil, fmt.Errorf("ws frame: consume log_id: %w", err)
}
f.LogID = v
b = b[m:]
case 3: // Service int32
if typ != protowire.VarintType {
return nil, fmt.Errorf("ws frame: field 3 expects varint, got %v", typ)
}
v, m := protowire.ConsumeVarint(b)
if err := protowire.ParseError(m); err != nil {
return nil, fmt.Errorf("ws frame: consume service: %w", err)
}
f.Service = int32(v)
b = b[m:]
case 4: // Method int32
if typ != protowire.VarintType {
return nil, fmt.Errorf("ws frame: field 4 expects varint, got %v", typ)
}
v, m := protowire.ConsumeVarint(b)
if err := protowire.ParseError(m); err != nil {
return nil, fmt.Errorf("ws frame: consume method: %w", err)
}
f.Method = int32(v)
b = b[m:]
case 5: // Headers (repeated)
if typ != protowire.BytesType {
return nil, fmt.Errorf("ws frame: field 5 expects bytes, got %v", typ)
}
hb, m := protowire.ConsumeBytes(b)
if err := protowire.ParseError(m); err != nil {
return nil, fmt.Errorf("ws frame: consume header: %w", err)
}
h, herr := unmarshalHeader(hb)
if herr != nil {
return nil, herr
}
f.Headers = append(f.Headers, h)
b = b[m:]
case 6: // PayloadEncoding string
if typ != protowire.BytesType {
return nil, fmt.Errorf("ws frame: field 6 expects bytes, got %v", typ)
}
s, m := protowire.ConsumeString(b)
if err := protowire.ParseError(m); err != nil {
return nil, fmt.Errorf("ws frame: consume payload_encoding: %w", err)
}
f.PayloadEncoding = s
b = b[m:]
case 7: // PayloadType string
if typ != protowire.BytesType {
return nil, fmt.Errorf("ws frame: field 7 expects bytes, got %v", typ)
}
s, m := protowire.ConsumeString(b)
if err := protowire.ParseError(m); err != nil {
return nil, fmt.Errorf("ws frame: consume payload_type: %w", err)
}
f.PayloadType = s
b = b[m:]
case 8: // Payload bytes
if typ != protowire.BytesType {
return nil, fmt.Errorf("ws frame: field 8 expects bytes, got %v", typ)
}
raw, m := protowire.ConsumeBytes(b)
if err := protowire.ParseError(m); err != nil {
return nil, fmt.Errorf("ws frame: consume payload: %w", err)
}
// Copy out so the Frame outlives the input buffer
// (ConsumeBytes returns a sub-slice).
f.Payload = append([]byte(nil), raw...)
b = b[m:]
case 9: // LogIDNew string
if typ != protowire.BytesType {
return nil, fmt.Errorf("ws frame: field 9 expects bytes, got %v", typ)
}
s, m := protowire.ConsumeString(b)
if err := protowire.ParseError(m); err != nil {
return nil, fmt.Errorf("ws frame: consume log_id_new: %w", err)
}
f.LogIDNew = s
b = b[m:]
default:
m := protowire.ConsumeFieldValue(num, typ, b)
if err := protowire.ParseError(m); err != nil {
return nil, fmt.Errorf("ws frame: skip unknown field %d: %w", num, err)
}
b = b[m:]
}
}
return f, nil
}
func unmarshalHeader(b []byte) (FrameHeader, error) {
var h FrameHeader
for len(b) > 0 {
num, typ, n := protowire.ConsumeTag(b)
if err := protowire.ParseError(n); err != nil {
return FrameHeader{}, fmt.Errorf("ws frame: header tag: %w", err)
}
b = b[n:]
switch num {
case 1:
if typ != protowire.BytesType {
return FrameHeader{}, fmt.Errorf("ws frame: header.key expects bytes, got %v", typ)
}
s, m := protowire.ConsumeString(b)
if err := protowire.ParseError(m); err != nil {
return FrameHeader{}, fmt.Errorf("ws frame: header.key: %w", err)
}
h.Key = s
b = b[m:]
case 2:
if typ != protowire.BytesType {
return FrameHeader{}, fmt.Errorf("ws frame: header.value expects bytes, got %v", typ)
}
s, m := protowire.ConsumeString(b)
if err := protowire.ParseError(m); err != nil {
return FrameHeader{}, fmt.Errorf("ws frame: header.value: %w", err)
}
h.Value = s
b = b[m:]
default:
m := protowire.ConsumeFieldValue(num, typ, b)
if err := protowire.ParseError(m); err != nil {
return FrameHeader{}, fmt.Errorf("ws frame: skip header field %d: %w", num, err)
}
b = b[m:]
}
}
return h, nil
}
// NewPingFrame builds the client-side keepalive frame. Lark's long
// connection uses an app-layer ping (binary Frame with type=ping),
// NOT a WebSocket protocol-level PING — gorilla's WriteControl pings
// would be ignored.
func NewPingFrame(serviceID int32) *Frame {
return &Frame{
Method: FrameMethodControl,
Service: serviceID,
Headers: []FrameHeader{{Key: FrameHeaderTypeKey, Value: FrameHeaderTypePing}},
}
}
// NewPongFrame builds the client-side response to a server-initiated
// ping. Lark may push ping frames at any cadence; we reply in kind.
func NewPongFrame(serviceID int32) *Frame {
return &Frame{
Method: FrameMethodControl,
Service: serviceID,
Headers: []FrameHeader{{Key: FrameHeaderTypeKey, Value: FrameHeaderTypePong}},
}
}
// NewAckFrame builds the ACK response for an inbound data frame.
// Per the SDK, the ACK reuses the inbound frame's Headers verbatim
// (so the server can correlate by message_id) and writes a JSON-
// encoded Response struct as the Payload.
//
// codeOK is true on successful dispatch (Response.code=200); false
// surfaces 500 to the server (it will retry the event).
func NewAckFrame(inbound *Frame, codeOK bool) *Frame {
code := 200
if !codeOK {
code = 500
}
payload := fmt.Sprintf(`{"code":%d,"headers":{},"data":""}`, code)
return &Frame{
Method: inbound.Method,
Service: inbound.Service,
Headers: inbound.Headers,
Payload: []byte(payload),
}
}
@@ -9,59 +9,50 @@ import (
db "github.com/multica-ai/multica/server/pkg/db/generated"
)
// LarkJSONFrameDecoder decodes the JSON envelopes Lark's
// event-subscription long-conn pushes over each WebSocket message.
// LarkJSONFrameDecoder decodes the JSON event payload Lark nests
// inside a long-conn data Frame. The outer binary Frame envelope
// (ws_frame.go) is stripped by the connector; the decoder only sees
// the bytes from Frame.Payload, which Lark formats as the standard
// event-subscription envelope: {schema, header, event}.
//
// The Lark open platform mixes a control surface (heartbeat / ack)
// with the actual event payload in a single channel; this decoder is
// the source of truth for which raw frames the dispatcher should ever
// see. It deliberately distinguishes three outcomes:
// Three outcomes:
//
// - (msg, true, nil) — a real `im.message.receive_v1` event that
// should land in chat_session. The Hub forwards it through the
// Dispatcher; the connector does not introspect further.
// - (zero, false, nil) — a recognized control frame (heartbeat,
// `frame_ack`, etc.) or a non-message event we don't yet handle.
// The connector drops these silently — they are expected traffic.
// - (zero, false, err) — malformed JSON or a recognized event we
// could not parse. The connector logs and drops the single frame,
// does NOT tear the WS down (one bad frame should not amplify
// into a reconnect storm).
// - (msg, true, nil) — `im.message.receive_v1` event. The Hub
// forwards through the Dispatcher.
// - (zero, false, nil) — heartbeat-shaped JSON or an event_type we
// don't yet handle (im.chat.access_event_v1, etc.). The connector
// drops these silently and still sends a 200 ACK to Lark so the
// server stops resending.
// - (zero, false, err) — malformed JSON or schema we couldn't
// parse. The connector logs + drops the single frame; the WS
// connection stays up because one bad payload shouldn't amplify
// into a reconnect storm.
//
// The Lark wire shape is the documented "p2_im_message_receive_v1"
// envelope; the broad outline is stable but Lark periodically adds
// fields. We extract only what the Dispatcher's InboundMessage needs
// and ignore the rest, which keeps the decoder forwards-compatible.
// The decoder is stateless and goroutine-safe — a single instance
// serves every supervisor goroutine.
type LarkJSONFrameDecoder struct{}
// NewLarkJSONFrameDecoder constructs the default Lark JSON decoder.
// The decoder holds no per-installation state, so a single instance
// can serve every supervisor goroutine.
func NewLarkJSONFrameDecoder() *LarkJSONFrameDecoder { return &LarkJSONFrameDecoder{} }
// Decode implements FrameDecoder.
func (d *LarkJSONFrameDecoder) Decode(raw []byte, inst db.LarkInstallation) (InboundMessage, bool, error) {
if len(raw) == 0 {
func (d *LarkJSONFrameDecoder) Decode(payload []byte, inst db.LarkInstallation) (InboundMessage, bool, error) {
if len(payload) == 0 {
return InboundMessage{}, false, nil
}
var env larkEventEnvelope
if err := json.Unmarshal(raw, &env); err != nil {
if err := json.Unmarshal(payload, &env); err != nil {
return InboundMessage{}, false, fmt.Errorf("envelope: %w", err)
}
// Heartbeats and acks come through with type=="heartbeat" or
// "frame_ack" depending on the protocol version. Any frame that
// is not an event_callback is dropped silently — including new
// kinds Lark might introduce that we have not taught the
// dispatcher about.
// Lark long-conn data frames are always v2 event envelopes
// (schema "2.0"). The legacy webhook v1 "type":"event_callback"
// shape is not used on long-conn — we accept it defensively in
// case Lark adds a back-compat mode, but the canonical path is
// schema-driven.
if env.Type != "" && env.Type != "event_callback" {
return InboundMessage{}, false, nil
}
// Validate the event header. The schema field discriminates
// payload shape; we only handle the documented v2.0 message-receive
// event in MVP, and drop everything else with a structured warning
// (returned to the caller as a non-error "ok=false").
if env.Header.EventType != "im.message.receive_v1" {
return InboundMessage{}, false, nil
}
@@ -84,21 +75,11 @@ func (d *LarkJSONFrameDecoder) Decode(raw []byte, inst db.LarkInstallation) (Inb
SenderOpenID: OpenID(evt.Sender.SenderID.OpenID),
}
// The body field for text messages carries a stringified JSON
// object like `{"text":"..."}`; we extract just the text so the
// dispatcher and the /issue parser see a plain string. Other
// message types (image, file, post) are not in MVP scope and
// surface as empty bodies — the dispatcher will still record the
// chat_message, just with no content.
switch evt.Message.MessageType {
case "text":
msg.Body = extractTextBody(evt.Message.Content)
}
// Group-mention discrimination: the WS payload includes a
// `mentions` array; we set AddressedToBot=true when the
// installation's bot_open_id appears there. Direct messages
// (p2p) ignore the field per InboundMessage doc.
if msg.ChatType == ChatTypeGroup {
msg.AddressedToBot = containsMention(evt.Message.Mentions, inst.BotOpenID)
}
@@ -107,13 +88,11 @@ func (d *LarkJSONFrameDecoder) Decode(raw []byte, inst db.LarkInstallation) (Inb
}
// larkEventEnvelope mirrors the outer JSON Lark wraps every push in.
// We keep Event as raw JSON so payload-shape changes for individual
// event types do not force a decoder revision.
type larkEventEnvelope struct {
Type string `json:"type"`
Header larkEventHeader `json:"header"`
Event json.RawMessage `json:"event"`
Schema string `json:"schema"`
Schema string `json:"schema"`
Type string `json:"type"`
Header larkEventHeader `json:"header"`
Event json.RawMessage `json:"event"`
}
type larkEventHeader struct {
@@ -125,7 +104,7 @@ type larkEventHeader struct {
}
// larkMessageReceiveEvent is the documented payload of
// im.message.receive_v1. Field names follow Lark's snake_case JSON.
// im.message.receive_v1.
type larkMessageReceiveEvent struct {
Sender struct {
SenderID struct {
@@ -137,13 +116,13 @@ type larkMessageReceiveEvent struct {
TenantKey string `json:"tenant_key"`
} `json:"sender"`
Message struct {
MessageID string `json:"message_id"`
ChatID string `json:"chat_id"`
ChatType string `json:"chat_type"`
MessageType string `json:"message_type"`
Content string `json:"content"`
Mentions []larkMention `json:"mentions"`
CreateTime string `json:"create_time"`
MessageID string `json:"message_id"`
ChatID string `json:"chat_id"`
ChatType string `json:"chat_type"`
MessageType string `json:"message_type"`
Content string `json:"content"`
Mentions []larkMention `json:"mentions"`
CreateTime string `json:"create_time"`
} `json:"message"`
}
@@ -165,9 +144,6 @@ func extractTextBody(content string) string {
Text string `json:"text"`
}
if err := json.Unmarshal([]byte(content), &doc); err != nil {
// Lark's text content is always a JSON object; if unmarshal
// fails we leave Body empty rather than dump the raw envelope
// into chat_session.
return ""
}
return doc.Text
@@ -0,0 +1,141 @@
package lark
import (
"bytes"
"testing"
)
// TestFrameRoundTripPreservesAllFields ensures every set field on the
// outbound Frame survives marshal+unmarshal. Field-tag mismatches
// against the SDK's pbbp2.Frame would silently corrupt frames on the
// wire — this test plus the explicit field numbers in ws_frame.go are
// the only checks that catch that.
func TestFrameRoundTripPreservesAllFields(t *testing.T) {
t.Parallel()
in := &Frame{
SeqID: 42,
LogID: 99,
Service: 7,
Method: FrameMethodData,
Headers: []FrameHeader{{Key: "type", Value: "event"}, {Key: "message_id", Value: "om-1"}},
PayloadEncoding: "json",
PayloadType: "im.message.receive_v1",
Payload: []byte(`{"schema":"2.0"}`),
LogIDNew: "log-new",
}
out, err := UnmarshalFrame(in.Marshal())
if err != nil {
t.Fatalf("UnmarshalFrame: %v", err)
}
if out.SeqID != in.SeqID || out.LogID != in.LogID || out.Service != in.Service || out.Method != in.Method {
t.Errorf("scalar fields differ: in=%+v out=%+v", in, out)
}
if len(out.Headers) != len(in.Headers) {
t.Fatalf("Headers len = %d; want %d", len(out.Headers), len(in.Headers))
}
for i, h := range out.Headers {
if h != in.Headers[i] {
t.Errorf("Headers[%d] = %+v; want %+v", i, h, in.Headers[i])
}
}
if out.PayloadEncoding != in.PayloadEncoding {
t.Errorf("PayloadEncoding = %q; want %q", out.PayloadEncoding, in.PayloadEncoding)
}
if out.PayloadType != in.PayloadType {
t.Errorf("PayloadType = %q; want %q", out.PayloadType, in.PayloadType)
}
if !bytes.Equal(out.Payload, in.Payload) {
t.Errorf("Payload = %q; want %q", string(out.Payload), string(in.Payload))
}
if out.LogIDNew != in.LogIDNew {
t.Errorf("LogIDNew = %q; want %q", out.LogIDNew, in.LogIDNew)
}
}
func TestFrameOmitsZeroFields(t *testing.T) {
t.Parallel()
// A minimal ping frame: Method=Control(0), Service=7, single
// header. Method=0 is proto3 default and skipped by the
// marshaller, so the encoded bytes contain only Service + Headers.
ping := NewPingFrame(7)
raw := ping.Marshal()
if len(raw) == 0 {
t.Fatal("ping marshal produced empty buffer")
}
out, err := UnmarshalFrame(raw)
if err != nil {
t.Fatalf("UnmarshalFrame: %v", err)
}
if out.Method != FrameMethodControl {
t.Errorf("Method = %d; want Control", out.Method)
}
if out.Service != 7 {
t.Errorf("Service = %d; want 7", out.Service)
}
if v := out.HeaderValue(FrameHeaderTypeKey); v != FrameHeaderTypePing {
t.Errorf("type header = %q; want ping", v)
}
if out.SeqID != 0 || out.LogID != 0 || len(out.Payload) != 0 {
t.Errorf("unset fields populated: %+v", out)
}
}
func TestNewAckFrameReusesInboundHeaders(t *testing.T) {
t.Parallel()
inbound := &Frame{
Method: FrameMethodData,
Service: 7,
Headers: []FrameHeader{
{Key: FrameHeaderTypeKey, Value: FrameHeaderTypeEvent},
{Key: FrameHeaderMessageIDKey, Value: "om-42"},
},
}
ack := NewAckFrame(inbound, true)
if ack.Method != inbound.Method || ack.Service != inbound.Service {
t.Errorf("ack method/service mismatch")
}
if len(ack.Headers) != len(inbound.Headers) {
t.Errorf("ack headers length mismatch")
}
if ack.HeaderValue(FrameHeaderMessageIDKey) != "om-42" {
t.Errorf("ack should echo message_id; got %q", ack.HeaderValue(FrameHeaderMessageIDKey))
}
if !contains(string(ack.Payload), `"code":200`) {
t.Errorf("ack payload missing code=200: %s", string(ack.Payload))
}
nack := NewAckFrame(inbound, false)
if !contains(string(nack.Payload), `"code":500`) {
t.Errorf("nack payload missing code=500: %s", string(nack.Payload))
}
}
func TestUnmarshalFrameRejectsTruncatedBuffer(t *testing.T) {
t.Parallel()
if _, err := UnmarshalFrame(nil); err == nil {
t.Error("expected error on empty buffer")
}
// Tag byte for field 1 (varint) with no following varint payload.
if _, err := UnmarshalFrame([]byte{0x08}); err == nil {
t.Error("expected error on truncated varint")
}
}
func TestUnmarshalFrameSkipsUnknownFields(t *testing.T) {
t.Parallel()
// Construct a buffer with one known field (Service=3, value=5)
// and one unknown field (number 31, varint=99). The unknown
// field MUST be skipped, not rejected.
buf := []byte{}
// field 3 (varint): tag = 3<<3|0 = 0x18, value = 5
buf = append(buf, 0x18, 0x05)
// field 31 (varint): tag = 31<<3|0 = 0xF8 0x01, value = 99 (0x63)
buf = append(buf, 0xF8, 0x01, 0x63)
f, err := UnmarshalFrame(buf)
if err != nil {
t.Fatalf("expected unknown field to be skipped, got error: %v", err)
}
if f.Service != 5 {
t.Errorf("Service = %d; want 5", f.Service)
}
}