package sitesd

import (
	"context"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"strings"
	"time"

	"github.com/bluesky-social/indigo/atproto/atcrypto"
	"github.com/bluesky-social/indigo/atproto/syntax"
	"tangled.org/core/sitesd/cache"
	"tangled.org/core/sitesd/cloudflare"
	"tangled.org/core/sitesd/config"
	"tangled.org/core/sitesd/models"
	"tangled.org/core/sitesd/reposync"
	"tangled.org/core/sitesd/sitefeed"
	"tangled.org/core/sitesd/sites"
	"tangled.org/core/sitesd/worker"
)

const (
	// deployTimeout bounds one deploy's R2 sync.
	deployTimeout = 10 * time.Minute
	// how long Stop lets in-flight deploys record their outcome
	deployShutdownGrace = 30 * time.Second
	// recordTimeout bounds the history write
	recordTimeout = 30 * time.Second
	// shutdownTimeout is the HTTP server drain budget at Stop.
	shutdownTimeout      = 15 * time.Second
	preflightTimeout     = 10 * time.Second
	internalWriteTimeout = 30 * time.Second
)

// Options are the runtime parameters of the site deploy executor, provided by
// the cmd/sitesd entrypoint (flags with env defaults).
type Options struct {
	Listen         string
	InternalListen string
	WorkerURL      string
	Audience       string
	Issuer         string
	KeyHex         string
	Knots          string
	Jetstream      string
	RedisAddr      string
	Dev            bool
}

// Serve runs the sitesd daemon until ctx is canceled: the push feed,
// the deploy HTTP surface (internal /deploy when InternalListen is set), did
// document, health.
func Serve(ctx context.Context, logger *slog.Logger, opts Options) error {
	var knots []string
	for _, knot := range strings.Split(opts.Knots, ",") {
		if knot = strings.TrimSpace(knot); knot != "" {
			knots = append(knots, knot)
		}
	}
	if opts.Jetstream == "" {
		return fmt.Errorf("no jetstream configured (SITESD_JETSTREAM_ENDPOINT); push-triggered deploys would never run")
	}

	runCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	key := opts.KeyHex
	if key == "" {
		key = os.Getenv("SITESD_PRIVATE_KEY")
	}
	priv, err := loadKey(key)
	if err != nil {
		return fmt.Errorf("loading P-256 key: %w", err)
	}

	cfg := &config.Config{
		Core: config.Core{Dev: opts.Dev},
		Cloudflare: config.Cloudflare{
			AccountId: os.Getenv("CLOUDFLARE_ACCOUNT_ID"),
			R2: config.R2Config{
				AccessKeyID:     os.Getenv("CLOUDFLARE_R2_ACCESS_KEY_ID"),
				SecretAccessKey: os.Getenv("CLOUDFLARE_R2_SECRET_ACCESS_KEY"),
				Bucket:          os.Getenv("CLOUDFLARE_R2_BUCKET"),
				BridgeURL:       os.Getenv("SITESD_R2_BRIDGE_URL"),
				BridgeToken:     os.Getenv("SITESD_R2_BRIDGE_TOKEN"),
			},
		},
	}

	cf, err := cloudflare.New(cfg)
	if err != nil {
		return fmt.Errorf("building cloudflare client: %w", err)
	}
	if !opts.Dev && (!cf.Enabled() || cfg.Cloudflare.R2.AccessKeyID == "" || cfg.Cloudflare.R2.SecretAccessKey == "") {
		return fmt.Errorf("cloudflare R2 requires CLOUDFLARE_ACCOUNT_ID and CLOUDFLARE_R2_* credentials")
	}
	if !cf.Enabled() {
		logger.Warn("cloudflare is not configured (CLOUDFLARE_ACCOUNT_ID, CLOUDFLARE_R2_*): every deploy is refused")
	}

	wrk := worker.New(opts.WorkerURL, opts.Issuer, opts.Audience, priv)

	var rdb *cache.Cache
	if opts.RedisAddr != "" {
		rdb = cache.New(opts.RedisAddr)
	} else {
		logger.Warn("no redis address (SITESD_REDIS): the jetstream cursor is not persisted, so a restart resumes live and the pushes it missed never deploy")
	}

	deployCtx, stopDeploys := context.WithCancel(context.Background())
	defer stopDeploys()

	locks := reposync.New()
	feed := sitefeed.New(rdb, cfg, cf, wrk, locks, deployCtx, knots, logger)
	if cf.Enabled() {
		if err := feed.Run(runCtx, opts.Jetstream); err != nil {
			return err
		}
	}

	mux := http.NewServeMux()
	srv := &server{
		cfg:       cfg,
		cf:        cf,
		wrk:       wrk,
		locks:     locks,
		logger:    logger,
		deployCtx: deployCtx,
		getCtx:    wrk.GetDeployContext,
		deploy:    sites.Deploy,
		record:    wrk.RecordDeploy,
	}
	mux.HandleFunc("GET /.well-known/did.json", srv.handleDidDoc(opts.Issuer, priv))
	mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
		w.WriteHeader(http.StatusOK)
	})

	logger.Info("sitesd listening", "addr", opts.Listen, "worker", opts.WorkerURL, "did", opts.Issuer)
	httpServer := &http.Server{
		Addr:              opts.Listen,
		Handler:           mux,
		ReadHeaderTimeout: 10 * time.Second,
		WriteTimeout:      10 * time.Minute,
		IdleTimeout:       2 * time.Minute,
	}
	serveErr := make(chan error, 2)
	go func() {
		if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			serveErr <- fmt.Errorf("public server: %w", err)
		}
	}()

	var internalServer *http.Server
	if opts.InternalListen != "" {
		// private-network only: the worker cannot sign service auth
		internalMux := http.NewServeMux()
		internalMux.HandleFunc("POST /deploy", srv.handleInternalDeploy)
		internalServer = &http.Server{
			Addr:              opts.InternalListen,
			Handler:           internalMux,
			ReadHeaderTimeout: 10 * time.Second,
			WriteTimeout:      internalWriteTimeout,
			IdleTimeout:       2 * time.Minute,
		}
		go func() {
			logger.Info("internal deploy trigger listening", "addr", opts.InternalListen)
			if err := internalServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
				serveErr <- fmt.Errorf("internal server: %w", err)
			}
		}()
	} else {
		logger.Warn("no internal listen address (SITESD_INTERNAL_ADDR): config-change and manual deploys have no endpoint")
	}

	var serveErrVal error
	select {
	case err := <-serveErr:
		serveErrVal = err
		cancel()
	case <-runCtx.Done():
	}
	logger.Info("shutting down", "reason", runCtx.Err())
	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), shutdownTimeout)
	defer shutdownCancel()
	if err := httpServer.Shutdown(shutdownCtx); err != nil {
		logger.Error("public server shutdown", "err", err)
	}
	if internalServer != nil {
		if err := internalServer.Shutdown(shutdownCtx); err != nil {
			logger.Error("internal server shutdown", "err", err)
		}
	}
	srv.drainDeploys(stopDeploys)
	return serveErrVal
}

func (s *server) drainDeploys(stop context.CancelFunc) {
	done := make(chan struct{})
	go func() {
		defer close(done)
		s.locks.Drain()
	}()
	select {
	case <-done:
	case <-time.After(deployShutdownGrace):
		s.logger.Warn("deploy drain grace elapsed, cancelling in-flight deploys", "grace", deployShutdownGrace)
		stop()
		<-done
	}
}

func loadKey(hexStr string) (*atcrypto.PrivateKeyP256, error) {
	if hexStr == "" {
		return nil, fmt.Errorf("no key provided: set SITESD_PRIVATE_KEY or -key (64 hex chars)")
	}
	raw, err := hex.DecodeString(strings.TrimSpace(hexStr))
	if err != nil {
		return nil, fmt.Errorf("decoding key hex: %w", err)
	}
	return atcrypto.ParsePrivateBytesP256(raw)
}

type server struct {
	cfg    *config.Config
	cf     *cloudflare.Client
	wrk    *worker.Client
	locks  *reposync.Registry
	logger *slog.Logger

	deployCtx context.Context

	getCtx func(context.Context, string) (*worker.DeployContext, error)
	deploy func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error
	record func(context.Context, *worker.DeployContext, string, string, string, string) error
}

type deployRequest struct {
	RepoDid string `json:"repoDid"`
	// an omitted trigger is a config-change deploy
	Trigger string `json:"trigger"`
}

type acceptedDeploy struct {
	dc      *worker.DeployContext
	trigger models.SiteDeployTrigger
	release func()
	done    func()
}

// ack 202 before deploy runs because worker wait_until expires in seconds while deploys take minutes
func (s *server) handleInternalDeploy(w http.ResponseWriter, r *http.Request) {
	var req deployRequest
	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&req); err != nil {
		writeErr(w, http.StatusBadRequest, "InvalidRequest", "request body must be JSON")
		return
	}
	if req.RepoDid == "" {
		writeErr(w, http.StatusBadRequest, "InvalidRequest", "missing repoDid")
		return
	}
	trigger := models.SiteDeployTriggerConfigChange
	if req.Trigger != "" {
		var ok bool
		if trigger, ok = models.ParseSiteDeployTrigger(req.Trigger); !ok {
			writeErr(w, http.StatusBadRequest, "InvalidRequest", "trigger must be one of config_change, push, manual")
			return
		}
	}
	if !s.cf.Enabled() {
		writeErr(w, http.StatusServiceUnavailable, "DeployUnavailable", "cloudflare is not configured")
		return
	}

	preflightCtx, preflightCancel := context.WithTimeout(r.Context(), preflightTimeout)
	defer preflightCancel()
	dc, err := s.getCtx(preflightCtx, req.RepoDid)
	if err != nil {
		s.logger.Error("internal deploy: deploy context unavailable", "repo", req.RepoDid, "err", err)
		writeErr(w, http.StatusBadGateway, "DeployContextUnavailable", err.Error())
		return
	}
	if dc == nil {
		writeErr(w, http.StatusUnprocessableEntity, "NoDeployContext", "no site config or active claim for this repo")
		return
	}

	// register before ack so draining waits for accepted work
	done, ok := s.locks.Begin()
	if !ok {
		writeErr(w, http.StatusServiceUnavailable, "ShuttingDown", "deploy executor is shutting down")
		return
	}
	release, ok := s.locks.TryAcquire()
	if !ok {
		done()
		writeErr(w, http.StatusTooManyRequests, "DeployAtCapacity", "every deploy slot is busy")
		return
	}

	go s.runDeploy(acceptedDeploy{dc: dc, trigger: trigger, release: release, done: done})

	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusAccepted)
	json.NewEncoder(w).Encode(map[string]string{"status": "accepted"})
}

func (s *server) runDeploy(a acceptedDeploy) {
	defer a.done()
	defer a.release()

	runCtx, cancel := context.WithTimeout(s.deployCtx, deployTimeout)
	defer cancel()

	dc := a.dc
	deploy := &models.SiteDeploy{
		RepoDid:   syntax.DID(dc.RepoDid),
		Branch:    dc.Branch,
		Dir:       dc.Dir,
		CommitSHA: "",
		Trigger:   a.trigger,
	}
	repo := &models.Repo{
		Did:     dc.OwnerDid,
		Name:    dc.Name,
		Rkey:    dc.Rkey,
		Knot:    dc.Knot,
		RepoDid: dc.RepoDid,
	}

	// push feed shares this lock, serializing against trigger deploys for the same repo
	unlock, lockErr := s.locks.Lock(runCtx, dc.RepoDid)
	var deployErr error
	if lockErr != nil {
		deployErr = fmt.Errorf("waiting for the repo lock: %w", lockErr)
	} else {
		defer unlock()
		deployErr = s.deploy(runCtx, s.cf, s.cfg, repo, dc.Branch, dc.Dir)
	}
	deploy.Status = models.SiteDeployStatusSuccess
	if deployErr != nil {
		deploy.Status = models.SiteDeployStatusFailure
		deploy.Error = deployErr.Error()
		s.logger.Error("internal deploy: failed", "repo", dc.RepoDid, "err", deployErr)
	}

	// outside the deploy's budget: timeouts and shutdowns still record
	recordCtx, recordCancel := context.WithTimeout(context.Background(), recordTimeout)
	defer recordCancel()
	if err := s.record(recordCtx, dc, deploy.CommitSHA, string(deploy.Status), string(deploy.Trigger), deploy.Error); err != nil {
		s.logger.Error("internal deploy: recording the outcome failed", "repo", dc.RepoDid, "status", deploy.Status, "err", err)
	}
}

func (s *server) handleDidDoc(issuer string, priv *atcrypto.PrivateKeyP256) http.HandlerFunc {
	pub, err := priv.PublicKey()
	multibase := ""
	if err == nil {
		multibase = pub.Multibase()
	}
	doc := map[string]any{
		"id": issuer,
		"verificationMethod": []map[string]string{
			{
				"id":                 issuer + "#atproto",
				"type":               "Multikey",
				"controller":         issuer,
				"publicKeyMultibase": multibase,
			},
		},
	}
	return func(w http.ResponseWriter, _ *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(doc)
	}
}

func writeErr(w http.ResponseWriter, status int, name, message string) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	json.NewEncoder(w).Encode(map[string]string{"error": name, "message": message})
}