event

var EventSubscribe = common.Shortcut{
	Service:     "event",
	Command:     "+subscribe",
	Description: "Subscribe to Lark events via WebSocket (NDJSON output)",
	Risk:        "read",
	Scopes:      []string{},
	AuthTypes:   []string{"bot"},
	Flags: []common.Flag{

		{Name: "output-dir", Desc: "write each event as a JSON file in this directory (default: stdout)"},
		{Name: "route", Type: "string_array", Desc: "regex-based event routing (e.g. --route '^im\\.message=dir:./im/' --route '^contact\\.=dir:./contacts/'); unmatched events fall through to --output-dir or stdout"},

		{Name: "compact", Type: "bool", Desc: "flat key-value output: extract text, strip noise fields"},
		{Name: "json", Type: "bool", Desc: "pretty-print JSON instead of NDJSON"},

		{Name: "event-types", Desc: "comma-separated event types to subscribe; only use when you do not need other events (omit for catch-all)"},
		{Name: "filter", Desc: "regex to further filter events by event_type"},

		{Name: "quiet", Type: "bool", Desc: "suppress stderr status messages"},
		{Name: "force", Type: "bool", Desc: "bypass single-instance lock (UNSAFE: server randomly splits events across connections, each instance only receives a subset)"},
	},
	DryRun: func(ctx context.Context, runtime *common.RuntimeContext) *common.DryRunAPI {
		eventTypesDisplay := "(catch-all)"
		if s := runtime.Str("event-types"); s != "" {
			eventTypesDisplay = s
		}
		filterDisplay := "(none)"
		if s := runtime.Str("filter"); s != "" {
			filterDisplay = s
		}
		outputDirDisplay := "(stdout)"
		if s := runtime.Str("output-dir"); s != "" {
			outputDirDisplay = s
		}
		routeDisplay := "(none)"
		if routes := runtime.StrArray("route"); len(routes) > 0 {
			routeDisplay = strings.Join(routes, "; ")
		}
		return common.NewDryRunAPI().
			Desc("Subscribe to Lark events via WebSocket (long-running)").
			Set("command", "event +subscribe").
			Set("app_id", runtime.Config.AppID).
			Set("event_types", eventTypesDisplay).
			Set("filter", filterDisplay).Set("output_dir", outputDirDisplay).
			Set("route", routeDisplay)
	},
	Execute: func(ctx context.Context, runtime *common.RuntimeContext) error {
		eventTypesStr := runtime.Str("event-types")
		filterStr := runtime.Str("filter")
		jsonFlag := runtime.Bool("json")
		compactFlag := runtime.Bool("compact")
		outputDir := runtime.Str("output-dir")
		quietFlag := runtime.Bool("quiet")
		routeSpecs := runtime.StrArray("route")
		forceFlag := runtime.Bool("force")

		if outputDir != "" {
			safePath, err := validate.SafeOutputPath(outputDir)
			if err != nil {
				return output.ErrValidation("unsafe output path: %s", err)
			}
			outputDir = safePath
		}

		errOut := runtime.IO().ErrOut
		out := runtime.IO().Out

		info := func(msg string) {
			if !quietFlag {
				fmt.Fprintln(errOut, msg)
			}
		}

		if !forceFlag {
			lock, err := lockfile.ForSubscribe(runtime.Config.AppID)
			if err != nil {
				return fmt.Errorf("failed to create lock: %w", err)
			}
			if err := lock.TryLock(); err != nil {
				return output.ErrValidation(
					"another event +subscribe instance is already running for app %s\n"+
						"  Only one subscriber per app is allowed to prevent competing consumers.\n"+
						"  Use --force to bypass this check.",
					runtime.Config.AppID,
				)
			}
			defer lock.Unlock()
		}

		eventTypeFilter := NewEventTypeFilter(eventTypesStr)
		regexFilter, err := NewRegexFilter(filterStr)
		if err != nil {
			return output.ErrValidation("invalid --filter regex: %s", filterStr)
		}
		var filterList []EventFilter
		if eventTypeFilter != nil {
			filterList = append(filterList, eventTypeFilter)
		}
		if regexFilter != nil {
			filterList = append(filterList, regexFilter)
		}
		filters := NewFilterChain(filterList...)

		router, err := ParseRoutes(routeSpecs)
		if err != nil {
			return output.ErrValidation("invalid --route: %v", err)
		}

		mode := TransformRaw
		if compactFlag {
			mode = TransformCompact
		}
		pipeline := NewEventPipeline(DefaultRegistry(), filters, PipelineConfig{
			Mode:      mode,
			JsonFlag:  jsonFlag,
			OutputDir: outputDir,
			Quiet:     quietFlag,
			Router:    router,
		}, out, errOut)

		if err := pipeline.EnsureDirs(); err != nil {
			return err
		}

		rawHandler := func(ctx context.Context, event *larkevent.EventReq) error {
			if event.Body == nil {
				return nil
			}
			var raw RawEvent
			if err := json.Unmarshal(event.Body, &raw); err != nil {
				output.PrintError(errOut, fmt.Sprintf("failed to parse event: %v", err))
				return nil
			}
			pipeline.Process(ctx, &raw)
			return nil
		}

		sdkLogger := &stderrLogger{w: errOut, quiet: quietFlag}

		eventDispatcher := dispatcher.NewEventDispatcher("", "")
		eventDispatcher.InitConfig(larkevent.WithLogger(sdkLogger))
		if eventTypeFilter != nil {
			for _, et := range eventTypeFilter.Types() {
				eventDispatcher.OnCustomizedEvent(et, rawHandler)
			}
		} else {
			for _, et := range commonEventTypes {
				eventDispatcher.OnCustomizedEvent(et, rawHandler)
			}
		}

		domain := lark.FeishuBaseUrl
		if runtime.Config.Brand == core.BrandLark {
			domain = lark.LarkBaseUrl
		}

		info(fmt.Sprintf("%sConnecting to Lark event WebSocket...%s", output.Cyan, output.Reset))
		if eventTypeFilter != nil {
			info(fmt.Sprintf("Listening for: %s%s%s", output.Green, strings.Join(eventTypeFilter.Types(), ", "), output.Reset))
		} else {
			info(fmt.Sprintf("Listening for %s%d common event types%s (catch-all mode)", output.Green, len(commonEventTypes), output.Reset))
			info(fmt.Sprintf("%sTip:%s use --event-types to listen for specific event types", output.Dim, output.Reset))
		}
		if regexFilter != nil {
			info(fmt.Sprintf("Filter: %s%s%s", output.Yellow, regexFilter.String(), output.Reset))
		}
		if router != nil {
			for _, spec := range routeSpecs {
				info(fmt.Sprintf("  Route: %s%s%s", output.Green, spec, output.Reset))
			}
		}

		cli := larkws.NewClient(runtime.Config.AppID, runtime.Config.AppSecret,
			larkws.WithEventHandler(eventDispatcher),
			larkws.WithDomain(domain),
			larkws.WithLogger(sdkLogger),
		)

		sigCh := make(chan os.Signal, 1)
		signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
		defer signal.Stop(sigCh)

		startErrCh := make(chan error, 1)
		go func() {
			startErrCh <- cli.Start(ctx)
		}()

		info(fmt.Sprintf("%s%sConnected.%s Waiting for events... (Ctrl+C to stop)", output.Bold, output.Green, output.Reset))

		select {
		case sig, ok := <-sigCh:
			if ok && sig != nil {
				info(fmt.Sprintf("\n%sReceived %s, shutting down...%s (received %s%d%s events)", output.Yellow, sig, output.Reset, output.Bold, pipeline.EventCount(), output.Reset))
			}
			return nil
		case err, ok := <-startErrCh:
			if !ok {
				return nil
			}
			if err != nil {
				return output.ErrNetwork("WebSocket connection failed: %v", err)
			}
			return nil
		}
	},
}