// Package server implements `dev --server`: it builds and launches agentteam-server
// plus the Vite web dev server, watches for Go
// source changes, and rebuilds on the fly.
package server

import (
	"context"
	"database/sql"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/dev/internal/build"
	"github.com/tidbcloud/AgentTeam/dev/internal/config"
	"github.com/tidbcloud/AgentTeam/dev/internal/env"
	"github.com/tidbcloud/AgentTeam/dev/internal/features"
	"github.com/tidbcloud/AgentTeam/dev/internal/watch"
	"github.com/tidbcloud/AgentTeam/x/proc"

	"github.com/fsnotify/fsnotify"
	_ "github.com/go-sql-driver/mysql"
)

const debounceDur = 2 * time.Second

// Service is a started dev service tracked by Runner.
type Service struct {
	Name   string
	Spec   build.Spec
	Runner *proc.Process
	ready  bool
}

// Runner owns the unified agentteam-server / web processes plus the file
// watcher for one `dev --server` invocation.
type Runner struct {
	repoRoot string
	opts     features.Options
	cfg      *config.Config
	mgr      *proc.Manager

	agentteamServer *Service
	web        *proc.Process

	webSpec  build.Spec
	agentteamSpec build.Spec
}

// New constructs a Runner. cfg must already have DSNs resolved by the TiDB
// provider.
func New(repoRoot string, opts features.Options, cfg *config.Config, mgr *proc.Manager) *Runner {
	return &Runner{
		repoRoot: repoRoot,
		opts:     opts,
		cfg:      cfg,
		mgr:      mgr,
		agentteamSpec: build.AgentTeamServer(repoRoot),
	}
}

// Run launches services and blocks until ctx is cancelled, watching for
// source changes that trigger rebuilds.
func (r *Runner) Run(ctx context.Context) error {
	slog.Info("== services ==")
	r.startServices(ctx)

	slog.Info("== watch ==")
	slog.Info("ready: watching for .go changes")
	return r.watchAgentTeam(ctx)
}

func (r *Runner) startServices(ctx context.Context) {
	r.agentteamServer = r.startService(ctx, r.agentteamSpec)
	if r.agentteamServer != nil && r.agentteamServer.ready {
		r.seedUser(ctx)
		r.web = r.startWeb(ctx)
	} else {
		slog.Info("agentteam-server is not ready; skipping web dev startup until the next successful agentteam-server restart")
	}
}

// seedUser creates the fixed-credential dev user only after explicit opt-in on
// a loopback-only server. Failures are logged but non-fatal.
func (r *Runner) seedUser(ctx context.Context) {
	url := os.Getenv(r.agentteamSpec.ReadyEnv)
	if url == "" {
		url = r.agentteamSpec.DefaultURL
	}
	seeded, err := seedDefaultUserIfEnabled(ctx, url)
	if err != nil {
		slog.Warn("seed default dev user failed; continuing", "error", err)
		return
	}
	if seeded {
		slog.Info("default dev user seeding enabled", "email", defaultDevUserEmail)
	}
}

func (r *Runner) startService(ctx context.Context, spec build.Spec) *Service {
	if !spec.Available() {
		return nil
	}
	if err := spec.Build(r.repoRoot); err != nil {
		slog.Error("initial build failed; waiting for changes", "service", spec.Name, "error", err)
		return nil
	}
	svc, err := r.launch(ctx, spec)
	if err != nil {
		slog.Error("start failed; waiting for changes", "service", spec.Name, "error", err)
		return nil
	}
	svc.ready = true
	return svc
}

func (r *Runner) launch(ctx context.Context, spec build.Spec) (*Service, error) {
	readyURL := os.Getenv(spec.ReadyEnv)
	if readyURL == "" {
		readyURL = spec.DefaultURL
	}
	p, err := r.mgr.Start(ctx, proc.Spec{
		Name: spec.Name,
		Args: []string{filepath.Join(r.repoRoot, spec.BinRel)},
		Dir:  spec.SourceDir,
	})
	if err != nil {
		return nil, err
	}
	if err := r.awaitReady(ctx, spec, p, readyURL); err != nil {
		r.mgr.Stop(p)
		return nil, err
	}
	return &Service{Name: spec.Name, Spec: spec, Runner: p}, nil
}

func (r *Runner) awaitReady(ctx context.Context, spec build.Spec, process *proc.Process, addr string) error {
	timeout := readyTimeoutFromEnv()
	deadline := time.Now().Add(timeout)
	url := readyURLForAddress(addr)
	client := &http.Client{Timeout: 3 * time.Second, Transport: &http.Transport{Proxy: nil}}
	checkDDL := spec.DDLAware && r.IsLocalTiDB()
	lastDDLLog := time.Now()
	lastProbe := "no probe attempted"
	slog.Info("waiting for ready", "service", spec.Name, "url", url, "timeout", timeout)

	for attempt := 1; ; attempt++ {
		if err := ctx.Err(); err != nil {
			return err
		}
		if err, exited := processExitError(process); exited {
			return fmt.Errorf("exited before ready: %w", err)
		}
		resp, err := client.Get(url)
		if err == nil {
			body, _ := io.ReadAll(io.LimitReader(resp.Body, 2048))
			resp.Body.Close()
			if resp.StatusCode == http.StatusOK {
				if err, exited := processExitError(process); exited {
					return fmt.Errorf("exited before ready: %w", err)
				}
				slog.Info("ready", "service", spec.Name)
				return nil
			}
			lastProbe = FormatHTTPProbe(resp)
			if trimmed := strings.TrimSpace(string(body)); trimmed != "" {
				lastProbe += ": " + trimmed
			}
		} else {
			lastProbe = err.Error()
		}

		ddlRunning := false
		if checkDDL {
			if jobs, ok := QueryDDLJobs(r.cfg); ok && len(jobs) > 0 {
				ddlRunning = true
				if time.Since(lastDDLLog) >= 30*time.Second {
					for _, job := range jobs {
						slog.Info("TiDB DDL in progress", "db", job.DBName, "table", job.Table, "type", job.Type, "state", job.State, "query", job.Query)
					}
					lastDDLLog = time.Now()
				}
			}
		}
		now := time.Now()
		deadline = nextReadyDeadline(deadline, now, timeout, ddlRunning)
		if !now.Before(deadline) {
			return fmt.Errorf("%s not ready after %s (last probe: %s)", spec.Name, timeout, lastProbe)
		}
		if attempt%10 == 0 {
			slog.Warn("still not ready", "service", spec.Name, "elapsed", attempt, "last_probe", lastProbe)
		}
		timer := time.NewTimer(time.Second)
		select {
		case <-ctx.Done():
			timer.Stop()
			return ctx.Err()
		case <-timer.C:
		}
	}
}

func readyTimeoutFromEnv() time.Duration {
	seconds, err := strconv.Atoi(strings.TrimSpace(os.Getenv("AGENTTEAM_DEV_READY_TIMEOUT")))
	if err != nil {
		seconds = 120
	}
	if seconds < 10 {
		seconds = 10
	}
	return time.Duration(seconds) * time.Second
}

func nextReadyDeadline(current, now time.Time, timeout time.Duration, ddlRunning bool) time.Time {
	if !ddlRunning {
		return current
	}
	extended := now.Add(timeout)
	if extended.After(current) {
		return extended
	}
	return current
}

func readyURLForAddress(addr string) string {
	switch {
	case strings.HasPrefix(addr, "http://"), strings.HasPrefix(addr, "https://"):
		return strings.TrimRight(addr, "/") + "/readyz"
	case strings.HasPrefix(addr, ":"):
		return "http://127.0.0.1" + addr + "/readyz"
	default:
		return "http://" + addr + "/readyz"
	}
}

func processExitError(process *proc.Process) (error, bool) {
	if process == nil || process.Done == nil {
		return nil, false
	}
	select {
	case <-process.Done:
		err := process.ExitErr()
		if err == nil {
			err = fmt.Errorf("process exited")
		}
		return err, true
	default:
		return nil, false
	}
}

// StopAll terminates every process owned by this Runner.
func (r *Runner) StopAll() {
	if r.agentteamServer != nil {
		r.mgr.Stop(r.agentteamServer.Runner)
		r.agentteamServer = nil
	}
	if r.web != nil {
		r.mgr.Stop(r.web)
		r.web = nil
	}
}

func (r *Runner) startWeb(ctx context.Context) *proc.Process {
	name, args := webDevCommand()
	env := replaceEnvList(os.Environ(), "AGENTTEAM_DEV_FEATURES", r.opts.Features.String())
	p, err := r.mgr.Start(ctx, proc.Spec{
		Name: "web",
		Args: append([]string{name}, args...),
		Dir:  filepath.Join(r.repoRoot, "web"),
		Env:  env,
	})
	if err != nil {
		slog.Warn("web dev failed to start", "error", err)
		return nil
	}
	return p
}

func webDevCommand() (string, []string) {
	if _, err := exec.LookPath("npm"); err == nil {
		return "npm", []string{"run", "dev"}
	}
	if _, err := exec.LookPath("bun"); err == nil {
		return "bun", []string{"run", "dev"}
	}
	return "npm", []string{"run", "dev"}
}

func replaceEnvList(env []string, name string, value string) []string {
	out := make([]string, 0, len(env)+1)
	replaced := false
	for _, entry := range env {
		current, _, ok := strings.Cut(entry, "=")
		if ok && current == name {
			if !replaced {
				out = append(out, name+"="+value)
				replaced = true
			}
			continue
		}
		out = append(out, entry)
	}
	if !replaced {
		out = append(out, name+"="+value)
	}
	return out
}

func (r *Runner) watchAgentTeam(ctx context.Context) error {
	w, err := fsnotify.NewWatcher()
	if err != nil {
		return err
	}
	defer w.Close()
	extraDirs, err := watch.ExtraFromEnv(r.repoRoot)
	if err != nil {
		return err
	}
	watch.AddSourceWatches(w, r.repoRoot, extraDirs)

	rebuildCh := make(chan watch.ChangeSet, 32)
	go r.rebuildWorker(ctx, rebuildCh)

	timer := time.NewTimer(0)
	if !timer.Stop() {
		<-timer.C
	}
	pending := watch.ChangeNone

	for {
		select {
		case <-ctx.Done():
			return nil
		case ev, ok := <-w.Events:
			if !ok {
				return nil
			}
			if !watch.IsSourceMutation(ev.Op) {
				continue
			}
			if ev.Has(fsnotify.Create) {
				if info, err := os.Stat(ev.Name); err == nil && info.IsDir() {
					w.Add(ev.Name)
				}
			}
			if target := watch.ChangeTarget(r.repoRoot, ev.Name, extraDirs); target != watch.ChangeNone && watch.IsRelevant(ev.Name) {
				pending = pending.Merge(target)
				slog.Info("changed", "path", relPath(r.repoRoot, ev.Name))
				resetTimer(timer, debounceDur)
			}
		case <-w.Errors:
		case <-timer.C:
			if pending != watch.ChangeNone {
				rebuildCh <- pending
				pending = watch.ChangeNone
			}
		}
	}
}

func (r *Runner) rebuildWorker(ctx context.Context, rebuildCh <-chan watch.ChangeSet) {
	for {
		select {
		case <-ctx.Done():
			return
		case target := <-rebuildCh:
			for {
				select {
				case next := <-rebuildCh:
					target = target.Merge(next)
				default:
					r.rebuildChangedServices(ctx, target)
					goto nextRequest
				}
			}
		nextRequest:
		}
	}
}

func (r *Runner) rebuildChangedServices(ctx context.Context, target watch.ChangeSet) {
	if target.Has(watch.ChangeAgentTeam) {
		r.rebuildService(ctx, &r.agentteamServer, r.agentteamSpec)
		if r.agentteamServer != nil && r.agentteamServer.ready {
			r.seedUser(ctx)
			if r.web == nil {
				r.web = r.startWeb(ctx)
			}
		}
	}
}

func (r *Runner) rebuildService(ctx context.Context, current **Service, spec build.Spec) {
	if current == nil || !spec.Available() {
		return
	}
	slog.Info("rebuilding...", "service", spec.Name)
	tmp, err := spec.BuildTemp(r.repoRoot)
	if err != nil {
		slog.Error("build failed; keeping current process and waiting for changes", "service", spec.Name, "error", err)
		return
	}
	if tmp == "" {
		return
	}
	if *current != nil {
		r.mgr.Stop((*current).Runner)
		*current = nil
	}
	if err := os.Rename(tmp, filepath.Join(r.repoRoot, spec.BinRel)); err != nil {
		_ = os.Remove(tmp)
		slog.Error("binary replace failed; waiting for changes", "service", spec.Name, "error", err)
		return
	}
	svc, err := r.launch(ctx, spec)
	if err != nil {
		slog.Error("start failed; waiting for changes", "service", spec.Name, "error", err)
		return
	}
	svc.ready = true
	*current = svc
}

func resetTimer(timer *time.Timer, d time.Duration) {
	if !timer.Stop() {
		select {
		case <-timer.C:
		default:
		}
	}
	timer.Reset(d)
}

func relPath(repoRoot, path string) string {
	if r, err := filepath.Rel(repoRoot, path); err == nil && !strings.HasPrefix(r, "..") {
		return r
	}
	return path
}

// IsLocalTiDB reports whether the active AGS DSN points at a local TiDB.
// Used to enable DDL-aware readiness waiting.
func (r *Runner) IsLocalTiDB() bool { return r.cfg.IsLocalTiDB() }

type DDLJob struct {
	DBName string
	Table  string
	Type   string
	State  string
	Query  string
}

// QueryDDLJobs returns running TiDB DDL jobs (for ready-wait logging).
func QueryDDLJobs(cfg *config.Config) ([]DDLJob, bool) {
	if cfg == nil || cfg.MeshXAGSDatabaseDSN == "" {
		return nil, false
	}
	connDSN := env.ReplaceDSNDatabase(cfg.MeshXAGSDatabaseDSN, "test")
	db, err := sql.Open("mysql", connDSN)
	if err != nil {
		return nil, false
	}
	defer db.Close()
	db.SetMaxOpenConns(1)
	db.SetConnMaxLifetime(5 * time.Second)

	rows, err := db.Query("ADMIN SHOW DDL JOBS WHERE STATE = 'running'")
	if err != nil {
		return nil, false
	}
	defer rows.Close()

	var jobs []DDLJob
	for rows.Next() {
		var job DDLJob
		var jobID int64
		var schemaState string
		var schemaID, tableID, rowCount int
		var createTime, startTime, endTime, comments, query sql.NullString
		if err := rows.Scan(
			&jobID, &job.DBName, &job.Table, &job.Type,
			&schemaState, &schemaID, &tableID, &rowCount,
			&createTime, &startTime, &endTime, &job.State, &comments, &query,
		); err != nil {
			continue
		}
		job.Query = query.String
		jobs = append(jobs, job)
	}
	return jobs, len(jobs) > 0
}

// FormatHTTPProbe converts an HTTP probe result into a one-line status string.
func FormatHTTPProbe(resp *http.Response) string {
	if resp == nil {
		return ""
	}
	return fmt.Sprintf("HTTP %s", resp.Status)
}