app.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"os"
	"path"
	"path/filepath"
	"sort"
	"strings"
	"time"

	"github.com/wailsapp/wails/v2/pkg/runtime"
	"github.com/yanbo92/topicconsole/internal/apperr"
	"github.com/yanbo92/topicconsole/internal/broker"
	"github.com/yanbo92/topicconsole/internal/config"
	"github.com/yanbo92/topicconsole/internal/discovery"
	"github.com/yanbo92/topicconsole/internal/profile"
	"github.com/yanbo92/topicconsole/internal/pulsaradmin"
	"github.com/yanbo92/topicconsole/internal/pulsarauth"
	"github.com/yanbo92/topicconsole/internal/pulsarconn"
	"github.com/yanbo92/topicconsole/internal/pulsarmeta"
	"github.com/yanbo92/topicconsole/internal/pulsarproduce"
	"github.com/yanbo92/topicconsole/internal/pulsartopic"
	"github.com/yanbo92/topicconsole/internal/schema"
	"github.com/yanbo92/topicconsole/internal/search"
	"github.com/yanbo92/topicconsole/internal/settingsbackup"
	"github.com/yanbo92/topicconsole/internal/stream"
	"github.com/yanbo92/topicconsole/internal/updater"
)

// App is the Wails-bound struct. All public methods become frontend RPCs.
// Keep methods thin — delegate to internal/ packages.
type App struct {
	ctx          context.Context
	version      string
	profileStore *profile.Store
	streamMgr    *stream.Manager
	searchMgr    *search.Manager
	metaScanMgr  *pulsarmeta.ScanManager
	metaCache    *broker.MetaCache
	avroCodec    *schema.Codec
	emitEvent    func(name string, data any)
	discoveryMgr *discovery.Manager
}

type resolvedProfileAuth struct {
	profile     profile.Profile
	clientToken string
	adminToken  string
}

func NewApp(version string) *App {
	return &App{version: version}
}

func (a *App) startup(ctx context.Context) {
	a.ctx = ctx

	store, err := profile.NewStore()
	if err != nil {
		slog.Error("failed to init profile store", "err", err)
		return
	}
	a.profileStore = store

	emit := func(name string, data any) {
		runtime.EventsEmit(ctx, name, data)
	}
	a.emitEvent = emit
	a.streamMgr = stream.NewManager(ctx, emit)
	a.searchMgr = search.NewManager(ctx, emit)
	a.metaScanMgr = pulsarmeta.NewScanManager(ctx, emit)

	a.metaCache = broker.NewMetaCache()
	a.avroCodec = schema.NewCodec()
	discoveryMgr, err := discovery.NewManager(store, a.avroCodec, emit)
	if err != nil {
		slog.Warn("failed to init discovery manager", "err", err)
	} else {
		a.discoveryMgr = discoveryMgr
	}

	go func() {
		time.Sleep(3 * time.Second)
		rel, err := updater.CheckLatest(a.version)
		if err != nil {
			slog.Warn("update check failed", "err", err)
			return
		}
		if rel != nil {
			runtime.EventsEmit(ctx, "app:update-available", rel)
		}
	}()
}

func (a *App) shutdown(_ context.Context) {
	a.searchMgr.StopAll()
	a.metaScanMgr.StopAll()
}

// ── App info & updates ────────────────────────────────────────────────────────

// GetAppVersion returns the current application version.
func (a *App) GetAppVersion() string {
	return a.version
}

// CheckForUpdates checks GitHub for a newer release.
// Returns the release info if an update is available, nil otherwise.
func (a *App) CheckForUpdates() (*updater.Release, error) {
	return updater.CheckLatest(a.version)
}

// OpenURL opens the given URL in the user's default browser.
func (a *App) OpenURL(url string) {
	runtime.BrowserOpenURL(a.ctx, url)
}

// ── Profile management ────────────────────────────────────────────────────────

func (a *App) ListProfiles() ([]profile.Profile, error) {
	return a.profileStore.List(), nil
}

func (a *App) GetActiveProfile() (*profile.Profile, error) {
	return a.profileStore.ActiveProfile()
}

func (a *App) CreateProfile(p profile.Profile) (profile.Profile, error) {
	if err := copyProfileCerts(&p); err != nil {
		return profile.Profile{}, fmt.Errorf("copy certs: %w", err)
	}
	return a.profileStore.Create(p)
}

func (a *App) DeleteProfile(id string) error {
	return a.profileStore.Delete(id)
}

func (a *App) ReorderProfiles(ids []string) error {
	return a.profileStore.Reorder(ids)
}

// ListDiscoveryConnectors returns locally available discovery connectors.
func (a *App) ListDiscoveryConnectors() ([]discovery.ConnectorInfo, error) {
	if a.discoveryMgr == nil {
		return nil, apperr.Validation("discovery", "discovery manager is not available")
	}
	return a.discoveryMgr.ListConnectors(), nil
}

// ListDiscoveryConnectorSources returns all editable on-disk HCL sources,
// including malformed files that are not part of the active connector set.
func (a *App) ListDiscoveryConnectorSources() ([]discovery.ConnectorSourceInfo, error) {
	if a.discoveryMgr == nil {
		return nil, apperr.Validation("discovery", "discovery manager is not available")
	}
	return a.discoveryMgr.ListConnectorSources()
}

// ReadDiscoveryConnectorSource reads one safely confined connector source.
func (a *App) ReadDiscoveryConnectorSource(filename string) (discovery.ConnectorSource, error) {
	if a.discoveryMgr == nil {
		return discovery.ConnectorSource{}, apperr.Validation("discovery", "discovery manager is not available")
	}
	return a.discoveryMgr.ReadConnectorSource(filename)
}

// ValidateDiscoveryConnectorSource validates editor text without mutating disk
// or the manager's active connector set.
func (a *App) ValidateDiscoveryConnectorSource(req discovery.ValidateConnectorSourceRequest) discovery.ConnectorValidation {
	if a.discoveryMgr == nil {
		return discovery.ConnectorValidation{Diagnostics: []discovery.ConnectorDiagnostic{{
			Severity: "error",
			Summary:  "Discovery unavailable",
			Detail:   "Discovery manager is not available.",
		}}}
	}
	return a.discoveryMgr.ValidateConnectorSource(req)
}

// SaveDiscoveryConnectorSource atomically saves a valid source and hot reloads
// the active connector set.
func (a *App) SaveDiscoveryConnectorSource(req discovery.SaveConnectorSourceRequest) (discovery.ConnectorSourceInfo, error) {
	if a.discoveryMgr == nil {
		return discovery.ConnectorSourceInfo{}, apperr.Validation("discovery", "discovery manager is not available")
	}
	return a.discoveryMgr.SaveConnectorSource(req)
}

// RestoreDiscoveryConnectorSource restores an embedded connector source and
// hot reloads the active connector set.
func (a *App) RestoreDiscoveryConnectorSource(filename string) (discovery.ConnectorSourceInfo, error) {
	if a.discoveryMgr == nil {
		return discovery.ConnectorSourceInfo{}, apperr.Validation("discovery", "discovery manager is not available")
	}
	return a.discoveryMgr.RestoreConnectorSource(filename)
}

// RunDiscoveryConnector runs a discovery connector and returns a redacted preview.
func (a *App) RunDiscoveryConnector(req discovery.RunConnectorRequest) (discovery.Preview, error) {
	if a.discoveryMgr == nil {
		return discovery.Preview{}, apperr.Validation("discovery", "discovery manager is not available")
	}
	return a.discoveryMgr.Run(req)
}

// SaveDiscoveryPreview persists a previously generated discovery preview.
func (a *App) SaveDiscoveryPreview(previewID string) (profile.Profile, error) {
	if a.discoveryMgr == nil {
		return profile.Profile{}, apperr.Validation("discovery", "discovery manager is not available")
	}
	return a.discoveryMgr.Save(previewID)
}

// RunDiscoveryRefresh reruns a profile's source connector and returns a redacted diff.
func (a *App) RunDiscoveryRefresh(req discovery.RefreshRequest) (discovery.RefreshPreview, error) {
	if a.discoveryMgr == nil {
		return discovery.RefreshPreview{}, apperr.Validation("discovery", "discovery manager is not available")
	}
	return a.discoveryMgr.RunRefresh(req)
}

// ApplyDiscoveryRefresh applies a previously generated discovery refresh diff.
func (a *App) ApplyDiscoveryRefresh(refreshID string) (profile.Profile, error) {
	if a.discoveryMgr == nil {
		return profile.Profile{}, apperr.Validation("discovery", "discovery manager is not available")
	}
	return a.discoveryMgr.ApplyRefresh(refreshID)
}

// ── Export/Import structures ──────────────────────────────────────────────────

type settingsExportDocument struct {
	Connections []exportProfile `json:"connections"`
}

type ExportSettingsResult struct {
	Cancelled bool `json:"cancelled"`
}

type ImportSettingsResult struct {
	Cancelled bool `json:"cancelled"`
	Imported  int  `json:"imported"`
	Skipped   int  `json:"skipped"`
}

type exportProfile struct {
	ID                 string                      `json:"id"`
	Name               string                      `json:"name"`
	Pulsar             profile.PulsarConfig        `json:"pulsar"`
	Auth               profile.AuthConfig          `json:"auth"`
	TLS                profile.TLSConfig           `json:"tls"`
	ActiveCredentialID string                      `json:"activeCredentialID,omitempty"`
	Token              string                      `json:"token,omitempty"`
	Credentials        []exportCredential          `json:"credentials,omitempty"`
	TopicGroups        []profile.TopicGroup        `json:"topicGroups,omitempty"`
	PinnedTopics       []string                    `json:"pinnedTopics,omitempty"`
	PinnedTenants      []string                    `json:"pinnedTenants,omitempty"`
	CustomTenants      []string                    `json:"customTenants,omitempty"`
	CustomNamespaces   []string                    `json:"customNamespaces,omitempty"`
	AvroSchemas        []profile.AvroSchemaMapping `json:"avroSchemas,omitempty"`
	Discovery          *profile.DiscoveryMetadata  `json:"discovery,omitempty"`
	CACertPEM          string                      `json:"caCertPEM,omitempty"`
	ClientCertPEM      string                      `json:"clientCertPEM,omitempty"`
	ClientKeyPEM       string                      `json:"clientKeyPEM,omitempty"`
}

type exportCredential struct {
	ID            string             `json:"id"`
	Name          string             `json:"name"`
	Auth          profile.AuthConfig `json:"auth"`
	AdminAuthMode string             `json:"adminAuthMode,omitempty"`
	AdminAuth     profile.AuthConfig `json:"adminAuth,omitempty"`
	Token         string             `json:"token,omitempty"`
	AdminToken    string             `json:"adminToken,omitempty"`
}

type avroSchemaResolvedEvent struct {
	Action     string `json:"action"`
	Topic      string `json:"topic"`
	SchemaPath string `json:"schemaPath"`
}

// ExportSettings saves all profiles to a user-chosen JSON file.
// When includeSecrets is false, tokens and saved discovery SSH auth are omitted.
func (a *App) ExportSettings(includeSecrets bool) (ExportSettingsResult, error) {
	return a.exportSettings(includeSecrets, runtime.SaveFileDialog, os.WriteFile)
}

func (a *App) exportSettings(
	includeSecrets bool,
	choosePath saveFileDialogFunc,
	writeFile writeFileFunc,
) (ExportSettingsResult, error) {
	path, err := choosePath(a.ctx, runtime.SaveDialogOptions{
		DefaultFilename: "topicconsole-backup.json",
		Filters: []runtime.FileFilter{
			{DisplayName: "JSON Files", Pattern: "*.json"},
		},
	})
	if err != nil {
		return ExportSettingsResult{}, err
	}
	if path == "" {
		return ExportSettingsResult{Cancelled: true}, nil
	}

	profiles := a.profileStore.List()
	export := settingsExportDocument{Connections: make([]exportProfile, 0, len(profiles))}

	for _, p := range profiles {
		ep := exportProfile{
			ID:                 p.ID,
			Name:               p.Name,
			Pulsar:             p.Pulsar,
			Auth:               p.Auth,
			TLS:                p.TLS,
			ActiveCredentialID: p.ActiveCredentialID,
			TopicGroups:        p.TopicGroups,
			PinnedTopics:       p.PinnedTopics,
			PinnedTenants:      p.PinnedTenants,
			CustomTenants:      p.CustomTenants,
			CustomNamespaces:   p.CustomNamespaces,
			AvroSchemas:        p.AvroSchemas,
			Discovery:          discoveryMetadataForExport(p.Discovery, includeSecrets),
		}
		ep.TLS.CACertPath = ""
		ep.TLS.ClientCertPath = ""
		ep.TLS.ClientKeyPath = ""
		if includeSecrets {
			ep.Token = p.Token
			if p.TLS.CACertPath != "" {
				if pem, err := os.ReadFile(p.TLS.CACertPath); err == nil {
					ep.CACertPEM = string(pem)
				} else {
					slog.Warn("export: read CA cert", "path", p.TLS.CACertPath, "err", err)
				}
			}
			if p.TLS.ClientCertPath != "" {
				if pem, err := os.ReadFile(p.TLS.ClientCertPath); err == nil {
					ep.ClientCertPEM = string(pem)
				} else {
					slog.Warn("export: read client cert", "path", p.TLS.ClientCertPath, "err", err)
				}
			}
			if p.TLS.ClientKeyPath != "" {
				if pem, err := os.ReadFile(p.TLS.ClientKeyPath); err == nil {
					ep.ClientKeyPEM = string(pem)
				} else {
					slog.Warn("export: read client key", "path", p.TLS.ClientKeyPath, "err", err)
				}
			}
		}
		for _, cred := range p.Credentials {
			ec := exportCredential{
				ID:            cred.ID,
				Name:          cred.Name,
				Auth:          cred.Auth,
				AdminAuthMode: cred.AdminAuthMode,
				AdminAuth:     cred.AdminAuth,
			}
			if includeSecrets {
				ec.Token = cred.Token
				ec.AdminToken = cred.AdminToken
			}
			ep.Credentials = append(ep.Credentials, ec)
		}
		export.Connections = append(export.Connections, ep)
	}

	data, err := json.MarshalIndent(export, "", "  ")
	if err != nil {
		return ExportSettingsResult{}, fmt.Errorf("marshal: %w", err)
	}
	if err := writeFile(path, data, 0o600); err != nil {
		return ExportSettingsResult{}, err
	}
	return ExportSettingsResult{}, nil
}

type saveFileDialogFunc func(context.Context, runtime.SaveDialogOptions) (string, error)
type openDirectoryDialogFunc func(context.Context, runtime.OpenDialogOptions) (string, error)
type writeFileFunc func(string, []byte, os.FileMode) error

type MessageExportFile struct {
	Filename string `json:"filename"`
	Content  string `json:"content"`
}

// SaveMessageExport writes serialized message data to a user-chosen file.
// A false result without an error means the user cancelled the dialog.
func (a *App) SaveMessageExport(topic, format, content string) (bool, error) {
	return saveMessageExport(a.ctx, topic, format, content, runtime.SaveFileDialog, writeMessageExportAtomic)
}

// SaveMessageExportFiles writes one serialized message per JSON file under a
// newly created export directory. A false result means the user cancelled.
func (a *App) SaveMessageExportFiles(topic string, files []MessageExportFile) (bool, error) {
	return saveMessageExportFiles(a.ctx, topic, files, runtime.OpenDirectoryDialog, os.WriteFile)
}

func saveMessageExport(
	ctx context.Context,
	topic string,
	format string,
	content string,
	choosePath saveFileDialogFunc,
	writeFile writeFileFunc,
) (bool, error) {
	filter, err := messageExportFilter(format)
	if err != nil {
		return false, err
	}

	selectedPath, err := choosePath(ctx, runtime.SaveDialogOptions{
		DefaultFilename: fmt.Sprintf("%s-%d.%s", messageExportBaseName(topic), time.Now().UnixMilli(), format),
		Title:           "Save Message Export",
		Filters:         []runtime.FileFilter{filter},
	})
	if err != nil {
		return false, fmt.Errorf("open message export dialog: %w", err)
	}
	if selectedPath == "" {
		return false, nil
	}

	if err := writeFile(selectedPath, []byte(content), 0o600); err != nil {
		return false, fmt.Errorf("write message export: %w", err)
	}
	return true, nil
}

func messageExportFilter(format string) (runtime.FileFilter, error) {
	switch format {
	case "jsonl":
		return runtime.FileFilter{DisplayName: "JSONL Files", Pattern: "*.jsonl"}, nil
	case "csv":
		return runtime.FileFilter{DisplayName: "CSV Files", Pattern: "*.csv"}, nil
	default:
		return runtime.FileFilter{}, fmt.Errorf("unsupported message export format %q", format)
	}
}

func saveMessageExportFiles(
	ctx context.Context,
	topic string,
	files []MessageExportFile,
	chooseDirectory openDirectoryDialogFunc,
	writeFile writeFileFunc,
) (bool, error) {
	if err := validateMessageExportFiles(files); err != nil {
		return false, err
	}

	parentDirectory, err := chooseDirectory(ctx, runtime.OpenDialogOptions{
		Title:                "Select Message Export Folder",
		CanCreateDirectories: true,
	})
	if err != nil {
		return false, fmt.Errorf("open message export directory dialog: %w", err)
	}
	if parentDirectory == "" {
		return false, nil
	}

	exportDirectory, err := createMessageExportDirectory(
		parentDirectory,
		messageExportBaseName(topic),
		time.Now().UnixMilli(),
	)
	if err != nil {
		return false, fmt.Errorf("create message export directory: %w", err)
	}

	for _, file := range files {
		filename := filepath.Join(exportDirectory, file.Filename)
		if err := writeFile(filename, []byte(file.Content), 0o600); err != nil {
			if cleanupErr := os.RemoveAll(exportDirectory); cleanupErr != nil {
				return false, fmt.Errorf("write message export file %q: %w; cleanup: %v", file.Filename, err, cleanupErr)
			}
			return false, fmt.Errorf("write message export file %q: %w", file.Filename, err)
		}
	}

	return true, nil
}

func validateMessageExportFiles(files []MessageExportFile) error {
	if len(files) == 0 {
		return fmt.Errorf("message export files must not be empty")
	}

	seen := make(map[string]struct{}, len(files))
	for _, file := range files {
		filename := file.Filename
		if filename == "" || filename != filepath.Base(filename) || strings.ContainsAny(filename, `/\`) {
			return fmt.Errorf("invalid message export filename %q", filename)
		}
		if filepath.Ext(filename) != ".json" {
			return fmt.Errorf("message export filename must end in .json: %q", filename)
		}
		for _, char := range filename {
			if char < 32 || strings.ContainsRune(`<>:"|?*`, char) {
				return fmt.Errorf("invalid message export filename %q", filename)
			}
		}

		key := strings.ToLower(filename)
		if _, exists := seen[key]; exists {
			return fmt.Errorf("duplicate message export filename %q", filename)
		}
		seen[key] = struct{}{}
	}
	return nil
}

func createMessageExportDirectory(parentDirectory, topicBase string, timestamp int64) (string, error) {
	base := fmt.Sprintf("%s-%d", topicBase, timestamp)
	for suffix := 1; ; suffix++ {
		name := base
		if suffix > 1 {
			name = fmt.Sprintf("%s-%d", base, suffix)
		}
		directory := filepath.Join(parentDirectory, name)
		if err := os.Mkdir(directory, 0o700); err == nil {
			return directory, nil
		} else if !os.IsExist(err) {
			return "", err
		}
	}
}

func messageExportBaseName(topic string) string {
	base := path.Base(strings.TrimSpace(topic))
	if base == "." || base == "/" || base == "" {
		return "messages"
	}

	base = strings.NewReplacer(
		"<", "-",
		">", "-",
		":", "-",
		`"`, "-",
		"/", "-",
		`\`, "-",
		"|", "-",
		"?", "-",
		"*", "-",
	).Replace(base)
	base = strings.Trim(base, " .")
	if base == "" {
		return "messages"
	}
	return base
}

func discoveryMetadataForExport(in *profile.DiscoveryMetadata, includeSecrets bool) *profile.DiscoveryMetadata {
	if in == nil {
		return nil
	}
	out := *in
	if in.Inputs != nil {
		out.Inputs = make(map[string]string, len(in.Inputs))
		for k, v := range in.Inputs {
			out.Inputs[k] = v
		}
	}
	if !includeSecrets {
		out.SSH = nil
	}
	return &out
}

// ImportSettings reads profiles from a user-chosen JSON file and merges them
// (adds profiles not already present by ID), restoring exported secrets.
// SelectCertificateFile opens a native file dialog for selecting certificate/key files.
func (a *App) SelectCertificateFile() (string, error) {
	return runtime.OpenFileDialog(a.ctx, runtime.OpenDialogOptions{
		Title: "Select Certificate File",
		Filters: []runtime.FileFilter{
			{DisplayName: "Certificates (*.pem, *.crt, *.cert, *.key)", Pattern: "*.pem;*.crt;*.cert;*.key"},
			{DisplayName: "All Files", Pattern: "*"},
		},
	})
}

// PickAvroSchemaFile opens a native file dialog for selecting local Avro schemas.
func (a *App) PickAvroSchemaFile() (string, error) {
	selectedPath, err := runtime.OpenFileDialog(a.ctx, runtime.OpenDialogOptions{
		Title: "Select Avro Schema",
		Filters: []runtime.FileFilter{
			{DisplayName: "Avro Schemas (*.avsc, *.avro)", Pattern: "*.avsc;*.avro"},
			{DisplayName: "All Files", Pattern: "*"},
		},
	})
	if err != nil || strings.TrimSpace(selectedPath) == "" {
		return selectedPath, err
	}

	configPath, _ := a.profileConfigPath()
	executablePath, _ := os.Executable()
	return relativeAvroSchemaPickerPath(selectedPath, configPath, executablePath), nil
}

// ValidateAvroSchemaMapping validates one local Avro schema mapping before it is saved.
func (a *App) ValidateAvroSchemaMapping(profileID string, topicPattern string, schemaPath string) error {
	profileID = strings.TrimSpace(profileID)
	if a.profileStore != nil && profileID != "" {
		if _, err := a.profileStore.Get(profileID); err != nil {
			return err
		}
	}
	if err := validateAvroSchemaTopicPattern(topicPattern); err != nil {
		return apperr.Validation("topic pattern", err.Error())
	}

	schemaPath = strings.TrimSpace(schemaPath)
	if schemaPath == "" {
		return apperr.Required("schema path")
	}
	resolvedPath, err := a.resolveLocalAvroSchemaPath(schemaPath)
	if err != nil {
		return apperr.Validation("avro schema", err.Error())
	}

	codec := a.avroCodec
	if codec == nil {
		codec = schema.NewCodec()
	}
	if err := codec.Validate(resolvedPath); err != nil {
		return apperr.Validation("avro schema", err.Error())
	}
	return nil
}

// Emits "profiles:imported" when at least one profile is imported.
func (a *App) ImportSettings() (ImportSettingsResult, error) {
	return a.importSettings(runtime.OpenFileDialog, func(name string, data any) {
		runtime.EventsEmit(a.ctx, name, data)
	})
}

func (a *App) importSettings(
	openFileDialog func(context.Context, runtime.OpenDialogOptions) (string, error),
	emitEvent func(string, any),
) (ImportSettingsResult, error) {
	path, err := openFileDialog(a.ctx, runtime.OpenDialogOptions{
		Filters: []runtime.FileFilter{
			{DisplayName: "JSON Files", Pattern: "*.json"},
		},
	})
	if err != nil {
		return ImportSettingsResult{}, err
	}
	if path == "" {
		return ImportSettingsResult{Cancelled: true}, nil
	}
	data, err := os.ReadFile(path)
	if err != nil {
		return ImportSettingsResult{}, fmt.Errorf("read file: %w", err)
	}

	document, err := settingsbackup.DecodeLegacy(data)
	if err != nil {
		return ImportSettingsResult{}, fmt.Errorf("parse settings: %w", err)
	}

	plan := settingsbackup.PlanImport(document, a.profileStore.List())
	certificateRoot := ""
	if settingsbackup.HasEmbeddedCertificates(plan.Connections) {
		configDirectory, err := config.Dir()
		if err != nil {
			return ImportSettingsResult{}, fmt.Errorf("restore certificates: %w", err)
		}
		certificateRoot = filepath.Join(configDirectory, "certs")
	}
	restored, err := settingsbackup.RestoreCertificates(certificateRoot, plan.Connections)
	if err != nil {
		return ImportSettingsResult{}, fmt.Errorf("restore certificates: %w", err)
	}
	if err := a.profileStore.ImportBatch(restored.Profiles); err != nil {
		if cleanupErr := restored.Cleanup(); cleanupErr != nil {
			return ImportSettingsResult{}, fmt.Errorf("import profiles: %w; cleanup certificates: %v", err, cleanupErr)
		}
		return ImportSettingsResult{}, fmt.Errorf("import profiles: %w", err)
	}

	result := ImportSettingsResult{Imported: len(restored.Profiles), Skipped: plan.Skipped}
	if result.Imported > 0 {
		emitEvent("profiles:imported", nil)
	}
	return result, nil
}

// RenameProfile updates only the name of a profile.
func (a *App) RenameProfile(id, name string) error {
	if strings.TrimSpace(name) == "" {
		return apperr.Required("name")
	}
	p, err := a.profileStore.Get(id)
	if err != nil {
		return err
	}
	p.Name = strings.TrimSpace(name)
	return a.profileStore.Update(*p)
}

// SwitchProfile stops all active sessions and clears caches,
// then switches the active profile.
func (a *App) SwitchProfile(id string) error {
	a.streamMgr.StopAll()
	a.searchMgr.StopAll()
	a.metaCache.InvalidateAll()

	if err := a.profileStore.SetActive(id); err != nil {
		return err
	}
	runtime.EventsEmit(a.ctx, "profile:switched", id)
	return nil
}

// ── Profile connection management ─────────────────────────────────────────────

// copyCertToAppData copies a certificate file into the app config directory.
// Returns the new path, or empty string if srcPath is empty.
func copyCertToAppData(profileID, filename, srcPath string) (string, error) {
	if srcPath == "" {
		return "", nil
	}
	dir, err := config.Dir()
	if err != nil {
		return srcPath, fmt.Errorf("config dir: %w", err)
	}
	if strings.HasPrefix(srcPath, dir) {
		return srcPath, nil // already inside app data
	}
	certsDir := filepath.Join(dir, "certs", profileID)
	if err := os.MkdirAll(certsDir, 0o755); err != nil {
		return srcPath, fmt.Errorf("create certs dir: %w", err)
	}
	data, err := os.ReadFile(srcPath)
	if err != nil {
		return srcPath, fmt.Errorf("read cert %s: %w", srcPath, err)
	}
	dst := filepath.Join(certsDir, filename)
	if err := os.WriteFile(dst, data, 0o600); err != nil {
		return srcPath, fmt.Errorf("write cert %s: %w", dst, err)
	}
	return dst, nil
}

// copyProfileCerts copies TLS certificate files into app data if they are external.
func copyProfileCerts(p *profile.Profile) error {
	var err error
	if p.TLS.CACertPath, err = copyCertToAppData(p.ID, "ca.pem", p.TLS.CACertPath); err != nil {
		return err
	}
	if p.TLS.ClientCertPath, err = copyCertToAppData(p.ID, "client-cert.pem", p.TLS.ClientCertPath); err != nil {
		return err
	}
	if p.TLS.ClientKeyPath, err = copyCertToAppData(p.ID, "client-key.pem", p.TLS.ClientKeyPath); err != nil {
		return err
	}
	return nil
}

func (a *App) UpdateProfile(p profile.Profile) error {
	if err := copyProfileCerts(&p); err != nil {
		return fmt.Errorf("copy certs: %w", err)
	}
	return a.profileStore.Update(p)
}

// AddProfileCredential adds a named credential to a profile.
func (a *App) AddProfileCredential(profileID string, cred profile.NamedCredential) (profile.NamedCredential, error) {
	return a.profileStore.AddCredential(profileID, cred)
}

// SetNamedCredentialPassword stores a named credential token in .topicconsole.json.
func (a *App) SetNamedCredentialPassword(profileID, credentialID, password string) error {
	return a.profileStore.SetCredentialToken(profileID, credentialID, password)
}

// SetNamedCredentialAdminPassword stores a named credential admin token in .topicconsole.json.
func (a *App) SetNamedCredentialAdminPassword(profileID, credentialID, password string) error {
	return a.profileStore.SetCredentialAdminToken(profileID, credentialID, password)
}

// DeleteProfileCredential removes a named credential from a profile.
func (a *App) DeleteProfileCredential(profileID, credentialID string) error {
	return a.profileStore.DeleteCredential(profileID, credentialID)
}

// SwitchProfileCredential stops all sessions for a profile, clears caches,
// sets the active credential, and emits an event.
func (a *App) SwitchProfileCredential(profileID, credentialID string) error {
	a.streamMgr.StopProfile(profileID)
	a.searchMgr.StopProfile(profileID)
	a.metaCache.InvalidateBroker(profileID)
	if err := a.profileStore.SetActiveCredential(profileID, credentialID); err != nil {
		return err
	}
	runtime.EventsEmit(a.ctx, "profile:credential-switched", map[string]string{
		"profileID":    profileID,
		"credentialID": credentialID,
	})
	return nil
}

// ClearActiveProfileCredential resets the profile to use its default auth settings.
func (a *App) ClearActiveProfileCredential(profileID string) error {
	a.streamMgr.StopProfile(profileID)
	a.searchMgr.StopProfile(profileID)
	a.metaCache.InvalidateBroker(profileID)
	return a.profileStore.ClearActiveCredential(profileID)
}

// adminReachable verifies the admin endpoint responds. Any HTTP response —
// including 401/403, e.g. a non-super-user token — counts as reachable: the
// service is up and the auth pipeline is wired. Only a transport-level
// failure (timeout/refused) is reported as an error.
func adminReachable(ctx context.Context, adminURL, adminToken string) error {
	_, err := pulsaradmin.NewClientWithToken(adminURL, adminToken).ListTenants(ctx)
	if err == nil || pulsaradmin.IsHTTPError(err) {
		return nil
	}
	return fmt.Errorf("reach Pulsar admin endpoint: %w", err)
}

// TestConnectionDirect tests a Pulsar connection using inline parameters,
// without requiring a saved broker. Useful for testing credentials before saving.
func (a *App) TestConnectionDirect(
	serviceURL string,
	adminURL string,
	tls profile.TLSConfig,
	auth profile.AuthConfig,
	token string,
	adminAuthMode string,
	adminAuth profile.AuthConfig,
	adminTokenSecret string,
) error {
	p := profile.Profile{
		Pulsar: profile.PulsarConfig{
			ServiceURL: serviceURL,
			AdminURL:   adminURL,
		},
		Auth: auth,
		TLS:  tls,
	}
	cfg := p.EffectivePulsarConfig()
	clientToken, adminToken, err := credentialAuthTokens(profile.NamedCredential{
		Auth:          auth,
		AdminAuthMode: adminAuthMode,
		AdminAuth:     adminAuth,
	}, token, adminTokenSecret)
	if err != nil {
		return err
	}
	if err := pulsarconn.TestConnection(a.ctx, pulsarconn.Config{ServiceURL: cfg.ServiceURL, AuthToken: clientToken}); err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	return adminReachable(ctx, cfg.AdminURL, adminToken)
}

// TestProfileConnection verifies the saved Pulsar service and admin URLs are reachable.
func (a *App) TestProfileConnection(profileID string) error {
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	if err := pulsarconn.TestConnection(a.ctx, pulsarconn.Config{ServiceURL: cfg.ServiceURL, AuthToken: resolved.clientToken}); err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	return adminReachable(ctx, cfg.AdminURL, resolved.adminToken)
}

// TestProfileConnectionWithOverrides verifies connectivity using the profile's
// saved active credential, but with the connection fields (service/admin URLs
// and TLS) replaced by the values currently entered in the edit dialog. This
// lets the user test unsaved edits before saving.
func (a *App) TestProfileConnectionWithOverrides(
	profileID string,
	serviceURL string,
	adminURL string,
	tls profile.TLSConfig,
) error {
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return err
	}
	p := profile.Profile{
		Pulsar: profile.PulsarConfig{
			ServiceURL: serviceURL,
			AdminURL:   adminURL,
		},
		TLS: tls,
	}
	cfg := p.EffectivePulsarConfig()
	if err := pulsarconn.TestConnection(a.ctx, pulsarconn.Config{ServiceURL: cfg.ServiceURL, AuthToken: resolved.clientToken}); err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	return adminReachable(ctx, cfg.AdminURL, resolved.adminToken)
}

// ── Pinned topics ─────────────────────────────────────────────────────────────

// PinTopic adds a topic to the profile's pinned list.
func (a *App) PinTopic(profileID, topic string) error {
	return a.profileStore.PinTopic(profileID, topic)
}

// UnpinTopic removes a topic from the profile's pinned list.
func (a *App) UnpinTopic(profileID, topic string) error {
	return a.profileStore.UnpinTopic(profileID, topic)
}

// PinTenant adds a tenant to the profile's pinned list so it sorts to the top.
func (a *App) PinTenant(profileID, tenant string) error {
	return a.profileStore.PinTenant(profileID, tenant)
}

// UnpinTenant removes a tenant from the profile's pinned list.
func (a *App) UnpinTenant(profileID, tenant string) error {
	return a.profileStore.UnpinTenant(profileID, tenant)
}

// ── Topic groups ──────────────────────────────────────────────────────────────

// SaveTopicGroup creates or updates a topic group for a profile.
func (a *App) SaveTopicGroup(profileID string, g profile.TopicGroup) error {
	return a.profileStore.SaveTopicGroup(profileID, g)
}

// DeleteTopicGroup removes a topic group from a profile.
func (a *App) DeleteTopicGroup(profileID, groupID string) error {
	return a.profileStore.DeleteTopicGroup(profileID, groupID)
}

// ── Broker metadata ───────────────────────────────────────────────────────────

type TenantListResult struct {
	Tenants []string `json:"tenants"`
	Warning string   `json:"warning,omitempty"`
}

type NamespaceListResult struct {
	Namespaces []broker.Namespace `json:"namespaces"`
	Warning    string             `json:"warning,omitempty"`
}

// ListTenants fetches live tenant names from Pulsar Admin, merged with any
// tenants the user added manually. When the admin API rejects the call for
// this token (e.g. a non-super-user token can't list all tenants), the
// manually-added tenants are returned alone so the tree is still usable.
func (a *App) ListTenants(profileID string) (TenantListResult, error) {
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return TenantListResult{}, err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	tenants, err := pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken).ListTenants(ctx)
	return buildTenantListResult(tenants, err, resolved.profile.CustomTenants, pinnedTenantSet(resolved.profile.PinnedTenants))
}

func buildTenantListResult(apiResult []string, apiErr error, custom []string, pinned map[string]bool) (TenantListResult, error) {
	tenants, err := mergeTenantList(apiResult, apiErr, custom, pinned)
	if apiErr != nil {
		if err != nil {
			tenants = []string{}
		}
		return TenantListResult{Tenants: tenants, Warning: apiErr.Error()}, nil
	}
	return TenantListResult{Tenants: tenants}, err
}

// mergeTenantList merges live tenants with a profile's manually-added custom
// tenants. When the admin API is unavailable to this token (apiErr != nil),
// custom tenants are served alone so a non-super-user can still browse the
// tenants they added by hand. The API error is returned only when there are no
// custom tenants to fall back on. Pinned tenants sort to the top.
func mergeTenantList(apiResult []string, apiErr error, custom []string, pinned map[string]bool) ([]string, error) {
	if apiErr != nil {
		if len(custom) > 0 {
			return dedupSortTenants(custom, pinned), nil
		}
		return nil, apiErr
	}
	merged := append([]string(nil), apiResult...)
	merged = append(merged, custom...)
	return dedupSortTenants(merged, pinned), nil
}

// dedupSortTenants trims and deduplicates tenant names, then sorts pinned
// tenants ahead of the rest. Each group stays alphabetical. A pinned name that
// is absent from the input is ignored, so pinning never injects a tenant the
// user cannot see.
func dedupSortTenants(in []string, pinned map[string]bool) []string {
	seen := make(map[string]bool, len(in))
	pinnedOut := make([]string, 0, len(in))
	rest := make([]string, 0, len(in))
	for _, t := range in {
		t = strings.TrimSpace(t)
		if t == "" || seen[t] {
			continue
		}
		seen[t] = true
		if pinned[t] {
			pinnedOut = append(pinnedOut, t)
		} else {
			rest = append(rest, t)
		}
	}
	sort.Strings(pinnedOut)
	sort.Strings(rest)
	return append(pinnedOut, rest...)
}

// pinnedTenantSet builds the lookup used to hoist pinned tenants to the top.
func pinnedTenantSet(pinned []string) map[string]bool {
	set := make(map[string]bool, len(pinned))
	for _, t := range pinned {
		t = strings.TrimSpace(t)
		if t != "" {
			set[t] = true
		}
	}
	return set
}

// AddCustomTenant adds a user-entered tenant to a profile so it appears in the
// tree even when the admin API rejects ListTenants for this token.
func (a *App) AddCustomTenant(profileID, tenant string) error {
	return a.profileStore.AddCustomTenant(profileID, tenant)
}

// RemoveCustomTenant removes a user-entered tenant from a profile.
func (a *App) RemoveCustomTenant(profileID, tenant string) error {
	return a.profileStore.RemoveCustomTenant(profileID, tenant)
}

// DeleteTenant deletes a tenant through Pulsar Admin.
func (a *App) DeleteTenant(profileID, tenant string) error {
	tenant = strings.TrimSpace(tenant)
	if tenant == "" {
		return apperr.Required("tenant")
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	if err := pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken).DeleteTenant(ctx, tenant); err != nil {
		return err
	}
	return a.profileStore.RemoveCustomTenant(profileID, tenant)
}

// ListNamespaces fetches live namespaces for a tenant and merges namespaces
// the user added manually. Manual namespaces remain available when the Admin
// token cannot list namespaces for the tenant.
func (a *App) ListNamespaces(profileID, tenant string) (NamespaceListResult, error) {
	tenant = strings.TrimSpace(tenant)
	if tenant == "" {
		return NamespaceListResult{}, apperr.Required("tenant")
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return NamespaceListResult{}, err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	fullNamespaces, err := pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken).ListNamespaces(ctx, tenant)
	return buildNamespaceListResult(tenant, fullNamespaces, err, resolved.profile.CustomNamespaces)
}

func buildNamespaceListResult(tenant string, apiResult []string, apiErr error, custom []string) (NamespaceListResult, error) {
	namespaces, err := mergeNamespaceList(tenant, apiResult, apiErr, custom)
	if apiErr != nil {
		if err != nil {
			namespaces = []broker.Namespace{}
		}
		return NamespaceListResult{Namespaces: namespaces, Warning: apiErr.Error()}, nil
	}
	return NamespaceListResult{Namespaces: namespaces}, err
}

func mergeNamespaceList(tenant string, apiResult []string, apiErr error, custom []string) ([]broker.Namespace, error) {
	fullNames := append([]string(nil), apiResult...)
	for _, fullName := range custom {
		nsTenant, _, err := pulsartopic.ParseNamespace(fullName)
		if err == nil && nsTenant == tenant {
			fullNames = append(fullNames, fullName)
		}
	}
	if apiErr != nil && len(fullNames) == 0 {
		return nil, apiErr
	}

	seen := make(map[string]bool, len(fullNames))
	namespaces := make([]broker.Namespace, 0, len(fullNames))
	for _, fullName := range fullNames {
		nsTenant, name, err := pulsartopic.ParseNamespace(fullName)
		if err != nil {
			return nil, err
		}
		if nsTenant != tenant || seen[fullName] {
			continue
		}
		seen[fullName] = true
		namespaces = append(namespaces, broker.Namespace{Tenant: nsTenant, Name: name, FullName: fullName})
	}
	sort.Slice(namespaces, func(i, j int) bool { return namespaces[i].Name < namespaces[j].Name })
	return namespaces, nil
}

// AddCustomNamespace adds a namespace to the tree without requiring list
// permission from Pulsar Admin.
func (a *App) AddCustomNamespace(profileID, tenant, namespace string) error {
	return a.profileStore.AddCustomNamespace(profileID, tenant, namespace)
}

// RemoveCustomNamespace removes a manually-added namespace from the tree.
func (a *App) RemoveCustomNamespace(profileID, tenant, namespace string) error {
	return a.profileStore.RemoveCustomNamespace(profileID, tenant, namespace)
}

// DeleteNamespace deletes a namespace through Pulsar Admin.
func (a *App) DeleteNamespace(profileID, tenant, namespace string) error {
	tenant = strings.TrimSpace(tenant)
	namespace = strings.TrimSpace(namespace)
	if tenant == "" {
		return apperr.Required("tenant")
	}
	if namespace == "" {
		return apperr.Required("namespace")
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	if err := pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken).DeleteNamespace(ctx, tenant, namespace); err != nil {
		return err
	}
	a.metaCache.InvalidateTopics(profileID, tenant, namespace)
	return a.profileStore.RemoveCustomNamespace(profileID, tenant, namespace)
}

// ListTopics fetches topics from the selected Pulsar tenant/namespace.
// Results are cached with a 30s TTL.
func (a *App) ListTopics(profileID, tenant, namespace string) ([]broker.Topic, error) {
	if cached, ok := a.metaCache.GetTopics(profileID, tenant, namespace); ok {
		return cached, nil
	}
	if strings.TrimSpace(tenant) == "" {
		return nil, apperr.Required("tenant")
	}
	if strings.TrimSpace(namespace) == "" {
		return nil, apperr.Required("namespace")
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return nil, err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	topics, err := pulsarmeta.ListTopics(ctx, pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken), tenant, namespace)
	if err != nil {
		return nil, err
	}
	a.metaCache.SetTopics(profileID, tenant, namespace, topics)
	return topics, nil
}

// InvalidateTopicsCache clears the topics cache for a selected namespace.
// Call before ListTopics to force a fresh fetch (e.g. on Refresh button).
func (a *App) InvalidateTopicsCache(profileID, tenant, namespace string) {
	a.metaCache.InvalidateTopics(profileID, tenant, namespace)
}

// ListSubscriptions fetches Pulsar subscriptions for the selected tenant/namespace.
func (a *App) ListSubscriptions(profileID, tenant, namespace string) ([]pulsarmeta.Subscription, error) {
	if strings.TrimSpace(tenant) == "" {
		return nil, apperr.Required("tenant")
	}
	if strings.TrimSpace(namespace) == "" {
		return nil, apperr.Required("namespace")
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return nil, err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	return pulsarmeta.ListSubscriptions(ctx, pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken), tenant, namespace)
}

// StartTopicScan discovers namespace topics immediately, then enriches
// partition metadata with bounded concurrency through Wails events.
func (a *App) StartTopicScan(profileID, tenant, namespace, requestID string) error {
	if strings.TrimSpace(tenant) == "" {
		return apperr.Required("tenant")
	}
	if strings.TrimSpace(namespace) == "" {
		return apperr.Required("namespace")
	}
	if strings.TrimSpace(requestID) == "" {
		return apperr.Required("request ID")
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	return a.metaScanMgr.StartTopicScan(requestID, pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken), tenant, namespace)
}

// StartSubscriptionScan scans topic stats with bounded concurrency and emits
// subscription batches as each group of topics completes.
func (a *App) StartSubscriptionScan(profileID, tenant, namespace, requestID string) error {
	if strings.TrimSpace(tenant) == "" {
		return apperr.Required("tenant")
	}
	if strings.TrimSpace(namespace) == "" {
		return apperr.Required("namespace")
	}
	if strings.TrimSpace(requestID) == "" {
		return apperr.Required("request ID")
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	return a.metaScanMgr.StartSubscriptionScan(requestID, pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken), tenant, namespace)
}

// StopMetadataScan cancels a topic or subscription metadata scan.
func (a *App) StopMetadataScan(requestID string) {
	a.metaScanMgr.Stop(requestID)
}

// ListTopicSubscriptions fetches Pulsar subscriptions for one exact topic.
func (a *App) ListTopicSubscriptions(profileID, topic string) ([]pulsarmeta.Subscription, error) {
	tenant, namespace, topicName, err := pulsartopic.ParseFullName(topic)
	if err != nil {
		return nil, apperr.Validation("topic", err.Error())
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return nil, err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	return pulsarmeta.ListTopicSubscriptions(ctx, pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken), tenant, namespace, topicName)
}

// SubscriptionExists reports whether a subscription exists on the exact topic.
func (a *App) SubscriptionExists(profileID, topic, subscriptionName string) (bool, error) {
	subscriptionName = strings.TrimSpace(subscriptionName)
	if subscriptionName == "" {
		return false, apperr.Required("subscription name")
	}
	tenant, namespace, topicName, err := pulsartopic.ParseFullName(topic)
	if err != nil {
		return false, apperr.Validation("topic", err.Error())
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return false, err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	return pulsarmeta.SubscriptionExists(ctx, pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken), tenant, namespace, topicName, subscriptionName)
}

// ClearSubscriptionBacklog skips all currently-backlogged messages for a topic subscription.
func (a *App) ClearSubscriptionBacklog(profileID, topic, subscriptionName string) error {
	subscriptionName = strings.TrimSpace(subscriptionName)
	if subscriptionName == "" {
		return apperr.Required("subscription name")
	}
	tenant, namespace, topicName, err := pulsartopic.ParseFullName(topic)
	if err != nil {
		return apperr.Validation("topic", err.Error())
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	return pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken).ClearSubscriptionBacklog(ctx, tenant, namespace, topicName, subscriptionName)
}

// DeleteSubscription removes a durable subscription from a topic.
func (a *App) DeleteSubscription(profileID, topic, subscriptionName string) error {
	subscriptionName = strings.TrimSpace(subscriptionName)
	if subscriptionName == "" {
		return apperr.Required("subscription name")
	}
	tenant, namespace, topicName, err := pulsartopic.ParseFullName(topic)
	if err != nil {
		return apperr.Validation("topic", err.Error())
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	return pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken).DeleteSubscription(ctx, tenant, namespace, topicName, subscriptionName)
}

// ── Producer ──────────────────────────────────────────────────────────────────

// ProduceMessage produces a single message to a Pulsar topic synchronously.
func (a *App) ProduceMessage(profileID string, req broker.ProduceRequest) error {
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return err
	}
	if _, _, _, err := pulsartopic.ParseFullName(req.Topic); err != nil {
		return apperr.Validation("topic", err.Error())
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	payload, schemaPath, err := a.resolveProducePayload(resolved.profile, req)
	if err != nil {
		return err
	}
	if err := pulsarproduce.ProduceMessageBytes(a.ctx, cfg.ServiceURL, resolved.clientToken, req, payload); err != nil {
		return err
	}
	a.notifyAvroSchemaResolved("produce", req.Topic, schemaPath)
	return nil
}

func (a *App) resolveProducePayload(p profile.Profile, req broker.ProduceRequest) ([]byte, string, error) {
	schemaPath, err := a.resolveReadableAvroSchemaPath(p, req.Topic)
	if err != nil {
		return nil, "", err
	}
	if schemaPath == "" {
		return []byte(req.Value), "", nil
	}

	codec := a.avroCodec
	if codec == nil {
		codec = schema.NewCodec()
	}
	payload, err := codec.EncodeJSON(schemaPath, req.Value)
	if err != nil {
		return nil, "", apperr.Validation("avro value", err.Error())
	}
	return payload, schemaPath, nil
}

// StartObserverAtTimestamp starts an observer at the given Unix millisecond publish timestamp.
func (a *App) StartObserverAtTimestamp(profileID, topic string, timestampMs int64, decodeOptions schema.DecodeOptions, receiverQueueSize int) (string, error) {
	if timestampMs <= 0 {
		return "", apperr.Positive("timestamp")
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return "", err
	}
	if _, _, _, err := pulsartopic.ParseFullName(topic); err != nil {
		return "", apperr.Validation("topic", err.Error())
	}
	schemaPath, err := a.resolveReadableAvroSchemaPath(resolved.profile, topic)
	if err != nil {
		return "", err
	}
	a.notifyAvroSchemaResolved("observe", topic, schemaPath)
	startTime := time.UnixMilli(timestampMs)
	return a.streamMgr.StartObserver(resolved.profile, resolved.clientToken, topic, stream.ObserverOpts{
		ReceiverQueueSize: receiverQueueSize,
		StartTime:         &startTime,
		Codec:             a.avroCodec,
		SchemaPath:        schemaPath,
		DecodeOptions:     &decodeOptions,
	})
}

// ── Topic management ──────────────────────────────────────────────────────────

// CreateTopic creates a new topic in the selected tenant/namespace.
func (a *App) CreateTopic(profileID, tenant, namespace string, req broker.CreateTopicRequest) error {
	if strings.TrimSpace(req.Name) == "" {
		return apperr.Required("topic name")
	}
	if strings.TrimSpace(tenant) == "" {
		return apperr.Required("tenant")
	}
	if strings.TrimSpace(namespace) == "" {
		return apperr.Required("namespace")
	}
	if req.Partitions <= 0 {
		return apperr.Positive("partitions")
	}
	if req.ReplicationFactor <= 0 {
		return apperr.Positive("replication factor")
	}
	if req.ReplicationFactor > 1 {
		return apperr.Validation("replication factor", "Pulsar topic replication is managed by namespace policies")
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	admin := pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken)
	topicName := pulsartopic.ShortName(req.Name)
	if req.Partitions > 1 {
		if err := admin.CreatePartitionedTopic(ctx, tenant, namespace, topicName, req.Partitions); err != nil {
			return err
		}
		a.metaCache.InvalidateTopics(profileID, tenant, namespace)
		return nil
	}
	if err := admin.CreateTopic(ctx, tenant, namespace, topicName); err != nil {
		return err
	}
	a.metaCache.InvalidateTopics(profileID, tenant, namespace)
	return nil
}

// DeleteTopic deletes a fully-qualified topic from the given profile.
func (a *App) DeleteTopic(profileID, topicName string) error {
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return err
	}
	tenant, namespace, shortName, err := pulsartopic.ParseFullName(topicName)
	if err != nil {
		return apperr.Validation("topic", err.Error())
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	admin := pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken)
	metadata, err := admin.GetPartitionedTopicMetadata(ctx, tenant, namespace, shortName)
	if err != nil && !pulsaradmin.IsHTTPStatus(err, 404) {
		return err
	}
	if err == nil && !metadata.Deleted && metadata.Partitions > 0 {
		if err := admin.DeletePartitionedTopic(ctx, tenant, namespace, shortName); err != nil {
			return err
		}
		a.metaCache.InvalidateTopics(profileID, tenant, namespace)
		return nil
	}
	if err := admin.DeleteTopic(ctx, tenant, namespace, shortName); err != nil {
		return err
	}
	a.metaCache.InvalidateTopics(profileID, tenant, namespace)
	return nil
}

// ── Stream sessions ───────────────────────────────────────────────────────────

// StartObserver starts reading a topic without creating a durable subscription.
// startPosition: "latest" (default) | "earliest"
// Returns the session ID to subscribe to "stream:<sessionID>" events.
func (a *App) StartObserver(profileID, topic, startPosition string, decodeOptions schema.DecodeOptions, receiverQueueSize int) (string, error) {
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return "", err
	}
	if _, _, _, err := pulsartopic.ParseFullName(topic); err != nil {
		return "", apperr.Validation("topic", err.Error())
	}
	schemaPath, err := a.resolveReadableAvroSchemaPath(resolved.profile, topic)
	if err != nil {
		return "", err
	}
	a.notifyAvroSchemaResolved("observe", topic, schemaPath)
	return a.streamMgr.StartObserver(resolved.profile, resolved.clientToken, topic, stream.ObserverOpts{
		ReceiverQueueSize: receiverQueueSize,
		StartPosition:     startPosition,
		Codec:             a.avroCodec,
		SchemaPath:        schemaPath,
		DecodeOptions:     &decodeOptions,
	})
}

// StartConsumer joins a Pulsar subscription and starts reading a topic.
// startPosition: "latest" (default) | "earliest" — initial position for new subscriptions.
// Returns the session ID to subscribe to "stream:<sessionID>" events.
func (a *App) StartConsumer(profileID, topic, subscriptionName, startPosition string, decodeOptions schema.DecodeOptions, receiverQueueSize int) (string, error) {
	if strings.TrimSpace(subscriptionName) == "" {
		return "", apperr.Required("subscription name")
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return "", err
	}
	if _, _, _, err := pulsartopic.ParseFullName(topic); err != nil {
		return "", apperr.Validation("topic", err.Error())
	}
	schemaPath, err := a.resolveReadableAvroSchemaPath(resolved.profile, topic)
	if err != nil {
		return "", err
	}
	a.notifyAvroSchemaResolved("consume", topic, schemaPath)
	return a.streamMgr.StartConsumer(resolved.profile, resolved.clientToken, topic, stream.ConsumerOpts{
		ReceiverQueueSize: receiverQueueSize,
		SubscriptionName:  subscriptionName,
		StartPosition:     startPosition,
		Codec:             a.avroCodec,
		SchemaPath:        schemaPath,
		DecodeOptions:     &decodeOptions,
	})
}

// StartConsumerAtTimestamp joins a Pulsar subscription and seeks it to the given Unix millisecond publish timestamp.
func (a *App) StartConsumerAtTimestamp(profileID, topic, subscriptionName string, timestampMs int64, decodeOptions schema.DecodeOptions, receiverQueueSize int) (string, error) {
	if strings.TrimSpace(subscriptionName) == "" {
		return "", apperr.Required("subscription name")
	}
	if timestampMs <= 0 {
		return "", apperr.Positive("timestamp")
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return "", err
	}
	if _, _, _, err := pulsartopic.ParseFullName(topic); err != nil {
		return "", apperr.Validation("topic", err.Error())
	}
	schemaPath, err := a.resolveReadableAvroSchemaPath(resolved.profile, topic)
	if err != nil {
		return "", err
	}
	a.notifyAvroSchemaResolved("consume", topic, schemaPath)
	startTime := time.UnixMilli(timestampMs)
	return a.streamMgr.StartConsumer(resolved.profile, resolved.clientToken, topic, stream.ConsumerOpts{
		ReceiverQueueSize: receiverQueueSize,
		SubscriptionName:  subscriptionName,
		StartTime:         &startTime,
		Codec:             a.avroCodec,
		SchemaPath:        schemaPath,
		DecodeOptions:     &decodeOptions,
	})
}

// CommitSession acknowledges all pending messages for a consumer session.
func (a *App) CommitSession(sessionID string) error {
	return a.streamMgr.CommitSession(a.ctx, sessionID)
}

// StopSession stops an active stream session.
func (a *App) StopSession(sessionID string) error {
	a.streamMgr.Stop(sessionID)
	return nil
}

// ── Search sessions ───────────────────────────────────────────────────────────

// StartSearch starts a topic search session. Returns the session ID.
// Frontend subscribes to "search:match:<sessionID>" and "search:progress:<sessionID>".
func (a *App) StartSearch(profileID string, req search.SearchRequest) (string, error) {
	if strings.TrimSpace(req.Topic) == "" {
		return "", apperr.Required("topic")
	}
	hasPattern := strings.TrimSpace(req.KeyPattern) != "" || strings.TrimSpace(req.ValuePattern) != ""
	hasTimeBound := req.TimestampFrom != nil || req.TimestampTo != nil
	if !hasPattern && !hasTimeBound {
		return "", apperr.Validation("search", "at least one key pattern, value pattern, or timestamp bound is required")
	}
	if req.TimestampFrom != nil && req.TimestampTo != nil && *req.TimestampTo < *req.TimestampFrom {
		return "", apperr.Validation("timestamp", "timestampTo must be greater than or equal to timestampFrom")
	}
	if req.MaxResults <= 0 {
		req.MaxResults = 1000
	}
	if req.MaxScan <= 0 {
		req.MaxScan = 1_000_000
	}

	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return "", err
	}
	if _, _, _, err := pulsartopic.ParseFullName(req.Topic); err != nil {
		return "", apperr.Validation("topic", err.Error())
	}
	schemaPath, err := a.resolveReadableAvroSchemaPath(resolved.profile, req.Topic)
	if err != nil {
		return "", err
	}
	a.notifyAvroSchemaResolved("search", req.Topic, schemaPath)
	return a.searchMgr.StartSearch(resolved.profile, resolved.clientToken, req, a.avroCodec, schemaPath, nil)
}

func (a *App) resolveReadableAvroSchemaPath(p profile.Profile, topic string) (string, error) {
	schemaPath := p.ResolveAvroSchemaPath(topic)
	if schemaPath == "" {
		return "", nil
	}

	resolvedPath, err := a.resolveLocalAvroSchemaPath(schemaPath)
	if err != nil {
		return "", apperr.Validation("avro schema", err.Error())
	}

	codec := a.avroCodec
	if codec == nil {
		codec = schema.NewCodec()
	}
	if err := codec.Validate(resolvedPath); err != nil {
		return "", apperr.Validation("avro schema", err.Error())
	}
	return resolvedPath, nil
}

func (a *App) resolveLocalAvroSchemaPath(schemaPath string) (string, error) {
	schemaPath = strings.TrimSpace(schemaPath)
	if schemaPath == "" || filepath.IsAbs(schemaPath) {
		return schemaPath, nil
	}

	configPath, err := a.profileConfigPath()
	if err != nil {
		return "", err
	}
	executablePath, err := os.Executable()
	if err != nil {
		return "", fmt.Errorf("resolve executable path: %w", err)
	}
	return resolveRelativeAvroSchemaPath(schemaPath, configPath, executablePath)
}

func (a *App) profileConfigPath() (string, error) {
	if a.profileStore != nil {
		return a.profileStore.Path(), nil
	}
	return config.ConnectionsFilePath()
}

func resolveRelativeAvroSchemaPath(schemaPath, configPath, executablePath string) (string, error) {
	candidates := make([]string, 0, 2)
	addCandidate := func(basePath string) {
		if strings.TrimSpace(basePath) == "" {
			return
		}
		candidate := filepath.Clean(filepath.Join(filepath.Dir(basePath), schemaPath))
		for _, existing := range candidates {
			if existing == candidate {
				return
			}
		}
		candidates = append(candidates, candidate)
	}
	addCandidate(configPath)
	addCandidate(executablePath)

	for _, candidate := range candidates {
		if _, err := os.Stat(candidate); err == nil {
			return candidate, nil
		} else if !os.IsNotExist(err) {
			return candidate, nil
		}
	}

	return "", fmt.Errorf("relative schema path %q not found; tried: %s", schemaPath, strings.Join(candidates, ", "))
}

func relativeAvroSchemaPickerPath(selectedPath, configPath, executablePath string) string {
	selectedPath = filepath.Clean(strings.TrimSpace(selectedPath))
	if selectedPath == "" || !filepath.IsAbs(selectedPath) {
		return selectedPath
	}

	for _, basePath := range []string{configPath, executablePath} {
		basePath = strings.TrimSpace(basePath)
		if basePath == "" {
			continue
		}
		baseDir := filepath.Dir(basePath)
		rel, err := filepath.Rel(baseDir, selectedPath)
		if err != nil || rel == "." || rel == "" {
			continue
		}
		if rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
			continue
		}
		return rel
	}

	return selectedPath
}

func validateAvroSchemaTopicPattern(pattern string) error {
	pattern = strings.TrimSpace(pattern)
	if pattern == "" {
		return fmt.Errorf("topic pattern is required")
	}
	if strings.Count(pattern, "*") == 0 {
		_, _, _, err := pulsartopic.ParseFullName(pattern)
		return err
	}
	if strings.Count(pattern, "*") != 1 {
		return fmt.Errorf("only one wildcard is supported")
	}

	const tenantWildcard = "://*/"
	if strings.Contains(pattern, tenantWildcard) {
		candidate := strings.Replace(pattern, tenantWildcard, "://tenant/", 1)
		tenant, _, topic, err := pulsartopic.ParseFullName(candidate)
		if err != nil {
			return err
		}
		if tenant != "tenant" || strings.TrimSpace(topic) == "*" {
			return fmt.Errorf("tenant wildcard must match persistent://*/namespace/topic")
		}
		return nil
	}

	if strings.HasSuffix(pattern, "/*") {
		candidate := strings.TrimSuffix(pattern, "/*") + "/topic"
		tenant, _, _, err := pulsartopic.ParseFullName(candidate)
		if err != nil {
			return err
		}
		if tenant == "*" {
			return fmt.Errorf("namespace wildcard must include a concrete tenant")
		}
		return nil
	}

	return fmt.Errorf("supported wildcards are persistent://tenant/namespace/* and persistent://*/namespace/topic")
}

func (a *App) notifyAvroSchemaResolved(action, topic, schemaPath string) {
	if strings.TrimSpace(schemaPath) == "" || a.emitEvent == nil {
		return
	}
	a.emitEvent("app:avro-schema-resolved", avroSchemaResolvedEvent{
		Action:     action,
		Topic:      topic,
		SchemaPath: schemaPath,
	})
}

// StopSearch stops an active search session.
func (a *App) StopSearch(sessionID string) error {
	a.searchMgr.Stop(sessionID)
	return nil
}

// ── Cluster metrics ───────────────────────────────────────────────────────────

// GetNamespaceTopicCount returns the topic count for the selected tenant/namespace.
// It is a lightweight, list-only call that avoids fetching per-topic stats.
func (a *App) GetNamespaceTopicCount(profileID, tenant, namespace string) (pulsarmeta.NamespaceTopicCount, error) {
	if strings.TrimSpace(tenant) == "" {
		return pulsarmeta.NamespaceTopicCount{}, apperr.Required("tenant")
	}
	if strings.TrimSpace(namespace) == "" {
		return pulsarmeta.NamespaceTopicCount{}, apperr.Required("namespace")
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return pulsarmeta.NamespaceTopicCount{}, err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	return pulsarmeta.GetNamespaceTopicCount(ctx, pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken), tenant, namespace)
}

// GetTopicStatsSummary returns producer, subscription, and consumer counts for
// the selected topic.
func (a *App) GetTopicStatsSummary(profileID, topic string) (pulsarmeta.TopicStatsSummary, error) {
	tenant, namespace, topicName, err := pulsartopic.ParseFullName(topic)
	if err != nil {
		return pulsarmeta.TopicStatsSummary{}, apperr.Validation("topic", err.Error())
	}
	resolved, err := a.resolveProfileAuth(profileID)
	if err != nil {
		return pulsarmeta.TopicStatsSummary{}, err
	}
	cfg := resolved.profile.EffectivePulsarConfig()
	ctx, cancel := context.WithTimeout(a.ctx, 10*time.Second)
	defer cancel()
	return pulsarmeta.GetTopicStatsSummary(ctx, pulsaradmin.NewClientWithToken(cfg.AdminURL, resolved.adminToken), tenant, namespace, topicName)
}

// ── Helpers ───────────────────────────────────────────────────────────────────

// resolveProfileAuth returns a copy of the profile with auth overridden by the
// active credential (if any) and resolved client/admin tokens.
// 1. ActiveCredentialID set → use that credential's auth + token
// 2. No active credential but credentials exist → auto-select first
// 3. No credentials → use the profile-level auth/token fields
func (a *App) resolveProfileAuth(profileID string) (resolvedProfileAuth, error) {
	p, err := a.profileStore.Get(profileID)
	if err != nil {
		return resolvedProfileAuth{}, err
	}
	pc := *p // work on a copy

	if pc.ActiveCredentialID != "" {
		for _, c := range pc.Credentials {
			if c.ID == pc.ActiveCredentialID {
				pc.Auth = c.Auth
				clientToken, adminToken, err := credentialAuthTokens(c, c.Token, c.AdminToken)
				if err != nil {
					return resolvedProfileAuth{}, err
				}
				return resolvedProfileAuth{profile: pc, clientToken: clientToken, adminToken: adminToken}, nil
			}
		}
	}

	if len(pc.Credentials) > 0 {
		c := pc.Credentials[0]
		pc.Auth = c.Auth
		clientToken, adminToken, err := credentialAuthTokens(c, c.Token, c.AdminToken)
		if err != nil {
			return resolvedProfileAuth{}, err
		}
		return resolvedProfileAuth{profile: pc, clientToken: clientToken, adminToken: adminToken}, nil
	}

	// No named credentials — use the profile-level auth settings.
	clientToken, err := pulsarauth.StaticToken(pc.Auth, pc.Token)
	if err != nil {
		return resolvedProfileAuth{}, err
	}
	return resolvedProfileAuth{profile: pc, clientToken: clientToken, adminToken: clientToken}, nil
}

func credentialAuthTokens(c profile.NamedCredential, clientSecret, adminSecret string) (string, string, error) {
	clientToken, err := pulsarauth.StaticToken(c.Auth, clientSecret)
	if err != nil {
		return "", "", err
	}

	switch normalizedAdminAuthMode(c.AdminAuthMode) {
	case profile.AdminAuthSame:
		return clientToken, clientToken, nil
	case profile.AdminAuthNone:
		return clientToken, "", nil
	case profile.AdminAuthToken:
		adminAuth := c.AdminAuth
		if strings.TrimSpace(adminAuth.Mode) == "" {
			adminAuth.Mode = pulsarauth.StaticTokenMode
		}
		adminToken, err := pulsarauth.StaticToken(adminAuth, adminSecret)
		if err != nil {
			return "", "", err
		}
		return clientToken, adminToken, nil
	default:
		return "", "", apperr.Validation("admin auth", "unsupported admin auth mode")
	}
}

func normalizedAdminAuthMode(mode string) string {
	switch strings.ToUpper(strings.TrimSpace(mode)) {
	case "", profile.AdminAuthSame:
		return profile.AdminAuthSame
	case profile.AdminAuthNone:
		return profile.AdminAuthNone
	case profile.AdminAuthToken:
		return profile.AdminAuthToken
	default:
		return mode
	}
}