larksuite__cli/cmd/event/bus.go

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

package event

import (
	"context"
	"os"
	"os/signal"
	"path/filepath"
	"syscall"

	"github.com/spf13/cobra"

	"github.com/larksuite/cli/errs"
	"github.com/larksuite/cli/internal/cmdutil"
	"github.com/larksuite/cli/internal/core"
	"github.com/larksuite/cli/internal/event"
	"github.com/larksuite/cli/internal/event/adapter/lark/websocket"
	"github.com/larksuite/cli/internal/event/adapter/localbus/transport"
	"github.com/larksuite/cli/internal/event/bus"
	"github.com/larksuite/cli/internal/event/catalog"
)

// NewCmdBus creates the hidden `event _bus` daemon subcommand, forked by the consume client; fork argv lives in consume/startup.go.
func NewCmdBus(f *cmdutil.Factory, snap *catalog.Snapshot) *cobra.Command {
	var domain string

	cmd := &cobra.Command{
		Use:    "_bus",
		Short:  "Internal event bus daemon (do not call directly)",
		Hidden: true,
		RunE: func(cmd *cobra.Command, args []string) error {
			cfg, err := f.Config()
			if err != nil {
				return err
			}

			// Sanitize AppID: an unsanitized value could escape events/ via ".." or separators.
			eventsDir := filepath.Join(core.GetConfigDir(), "events", event.SanitizeAppID(cfg.AppID))

			logger, err := bus.SetupBusLogger(eventsDir)
			if err != nil {
				return errs.NewInternalError(errs.SubtypeFileIO,
					"set up bus logger: %s", err).WithCause(err)
			}

			tr := transport.New()
			ingress := &websocket.FeishuSource{
				AppID:     cfg.AppID,
				AppSecret: cfg.AppSecret,
				Domain:    domain,
				Logger:    logger,
			}
			b := bus.NewBus(cfg.AppID, cfg.AppSecret, domain, tr, logger, snap, ingress)

			ctx, cancel := context.WithCancel(cmd.Context())
			defer cancel()

			sigCh := make(chan os.Signal, 1)
			signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
			defer signal.Stop(sigCh)
			go func() {
				select {
				case <-sigCh:
					cancel()
				case <-ctx.Done():
				}
			}()

			if err := b.Run(ctx); err != nil {
				if _, ok := errs.ProblemOf(err); ok {
					return err
				}
				return errs.NewInternalError(errs.SubtypeUnknown,
					"event bus daemon exited: %s", err).WithCause(err)
			}
			return nil
		},
	}

	cmd.Flags().StringVar(&domain, "domain", "", "API domain")
	_ = cmd.Flags().MarkHidden("domain")
	cmdutil.SetRisk(cmd, "write")

	return cmd
}