package cloud

import (
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/kerberos-io/agent/machinery/src/log"
	"github.com/kerberos-io/agent/machinery/src/models"
)

// tusResumableVersion is the tus protocol version implemented by this client.
const tusResumableVersion = "1.0.0"

// tusUploadPath is appended to the configured Kerberos Vault URI to reach the
// resumable upload endpoint. It mirrors how the legacy uploader appends
// "/storage".
const tusUploadPath = "/storage/tus/"

// tusResumeState is persisted in a sidecar file next to the agent data so an
// interrupted upload can be resumed across retries and even agent restarts.
type tusResumeState struct {
	UploadURL string `json:"upload_url"`
	VaultURI  string `json:"vault_uri"`
	Size      int64  `json:"size"`
}

// resumableUploadsEnabled reports whether the resumable (tus) upload path should
// be attempted. It is enabled by default and can be disabled (falling back to
// the legacy single POST) by setting AGENT_DISABLE_RESUMABLE_UPLOAD=true.
func resumableUploadsEnabled() bool {
	return os.Getenv("AGENT_DISABLE_RESUMABLE_UPLOAD") != "true"
}

// tusDefaultChunkSize is the number of bytes uploaded per PATCH request when no
// explicit size is configured. Splitting the upload into chunks keeps each HTTP
// request small enough for intermediary proxies/load balancers and checkpoints
// progress frequently, so an interruption resumes with minimal re-upload.
const tusDefaultChunkSize int64 = 8 << 20 // 8 MiB (>= S3 multipart minimum part size)

const tusProgressBucketPercent int64 = 10

// tusChunkSize returns the number of bytes to send per PATCH request. It
// defaults to tusDefaultChunkSize (8 MiB) and can be overridden with the
// AGENT_TUS_CHUNK_SIZE_BYTES environment variable. A value of 0 (or negative)
// disables chunking and sends the remaining bytes in a single PATCH.
func tusChunkSize() int64 {
	v := os.Getenv("AGENT_TUS_CHUNK_SIZE_BYTES")
	if v == "" {
		return tusDefaultChunkSize
	}
	n, err := strconv.ParseInt(v, 10, 64)
	if err != nil {
		return tusDefaultChunkSize
	}
	if n <= 0 {
		return 0 // chunking disabled: send everything in one PATCH
	}
	return n
}

func tusProgressBucket(offset, size int64) int64 {
	if size <= 0 {
		return 100
	}
	percent := (offset * 100) / size
	if percent > 100 {
		percent = 100
	}
	return percent / tusProgressBucketPercent
}

func logTusUploadProgress(label string, offset, size int64, loggedBucket *int64) {
	bucket := tusProgressBucket(offset, size)
	if bucket <= *loggedBucket {
		return
	}
	*loggedBucket = bucket
	percent := bucket * tusProgressBucketPercent
	if percent > 100 {
		percent = 100
	}
	log.Log.Infof("%s: resumable upload progress %d%% (%d/%d bytes)", label, percent, offset, size)
}

// tusHeaderFunc sets the authentication and routing headers required on every
// tus request for a particular upload target (Kerberos Vault directly, or
// Kerberos Hub which proxies to a vault). fileName is only meaningful on the
// creation request; it is empty on HEAD/PATCH/DELETE.
type tusHeaderFunc func(h http.Header, fileName string)

// runTusUpload performs a resumable (tus) upload of data/recordings/<fileName>
// to baseURL, sending target-specific authentication/routing headers via
// setHeaders on every request. It encapsulates the create/resume/chunk/finalize
// state machine shared by the Kerberos Vault (direct) and Kerberos Hub (proxied)
// upload paths.
//
// Return values:
//   - uploaded:  the recording was fully received and persisted by the server.
//   - responded: the server returned a definitive HTTP response (used by the
//     caller to advance its retry/secondary-failover policy).
//   - supported: the server exposes a tus endpoint. When false, the caller
//     should fall back to the legacy single-POST upload (older deployments).
//   - body:      a short message for logging.
func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tusHeaderFunc) (uploaded bool, responded bool, supported bool, body string, err error) {
	fullname := "data/recordings/" + fileName

	file, ferr := os.Open(fullname)
	if file != nil {
		defer file.Close()
	}
	if ferr != nil {
		msg := label + ": resumable upload failed, file doesn't exist anymore"
		log.Log.Info(msg)
		// The file is gone, so the legacy path cannot help either. Report it as
		// "supported" to avoid a pointless fallback attempt.
		return false, false, true, "", errors.New(msg)
	}

	info, serr := file.Stat()
	if serr != nil {
		return false, false, true, "", serr
	}
	size := info.Size()

	client := newVaultHTTPClient(0)
	client.CheckRedirect = func(req *http.Request, via []*http.Request) error {
		if len(via) == 0 {
			return nil
		}
		if req.URL.Host != via[0].URL.Host {
			for k := range req.Header {
				if strings.HasPrefix(http.CanonicalHeaderKey(k), "X-Kerberos-") {
					req.Header.Del(k)
				}
			}
		}
		return nil
	}

	sidecar := tusSidecarPath(fileName, slot)
	uploadURL := loadTusResumeState(sidecar, baseURL)

	const maxAttempts = 4
	restartedAfterComplete := false

	// lastStatus holds the HTTP status code of the most recent tus request. A
	// value of 0 means the request failed at the transport level (no HTTP
	// response at all, e.g. the vault was unreachable or the connection dropped
	// because the internet went down). It lets the final "gave up" return report
	// whether the vault actually answered, so the caller only advances its
	// retry/back-off policy on a definitive response and transient network errors
	// never consume the retry budget (matching the legacy single-POST behaviour).
	lastStatus := 0

	// highWaterOffset is the furthest server-acknowledged offset observed across
	// all attempts (via HEAD or PATCH). It lets the retry budget be refreshed only
	// on GENUINE net forward progress. Without it, a server that keeps resetting the
	// offset — e.g. a persistent 409 ERR_MISMATCHED_OFFSET where HEAD reports 0 again
	// while the first chunk still "succeeds" — would refresh the budget every attempt
	// and loop forever, wedging the upload worker on one recording and saturating the
	// uplink.
	highWaterOffset := int64(0)

	for attempt := 0; attempt < maxAttempts; attempt++ {
		// (1) Ensure we have an active upload URL, creating one if needed.
		if uploadURL == "" {
			created, status, cerr := tusCreate(client, baseURL, size, metadata, setHeaders, fileName)
			lastStatus = status
			if cerr != nil {
				if status == http.StatusNotFound || status == http.StatusMethodNotAllowed || status == http.StatusNotImplemented {
					// The vault does not implement tus; let the caller fall back.
					return false, false, false, "", cerr
				}
				log.Log.Info(label + ": resumable create failed, " + cerr.Error())
				tusBackoff(attempt)
				continue
			}
			uploadURL = created
			saveTusResumeState(sidecar, tusResumeState{UploadURL: uploadURL, VaultURI: baseURL, Size: size})
		}

		// (2) Query the current server-side offset.
		offset, status, herr := tusHead(client, uploadURL, setHeaders)
		lastStatus = status
		if herr != nil {
			if status == http.StatusNotFound || status == http.StatusGone {
				// The upload expired/was removed server-side; start over.
				removeTusResumeState(sidecar)
				uploadURL = ""
				continue
			}
			log.Log.Info(label + ": resumable head failed, " + herr.Error())
			tusBackoff(attempt)
			continue
		}

		// The furthest offset any previous attempt reached. If this attempt pushes
		// past it (via HEAD showing server-side progress or a successful PATCH) we made
		// genuine net progress and may refresh the retry budget; if not, a repeated
		// failure at the same spot must count against maxAttempts.
		startHighWater := highWaterOffset
		if offset > highWaterOffset {
			highWaterOffset = offset
		}

		// (3) All bytes are present but the upload was not finalized (e.g. the
		// completion hook failed). A completed tus upload cannot be re-finalized
		// with another PATCH, so delete it and re-upload to force a clean finalize.
		if offset >= size {
			if restartedAfterComplete {
				return false, true, true, "resumable finalize did not complete", errors.New(label + ": resumable finalize did not complete")
			}
			tusTerminate(client, uploadURL, setHeaders)
			removeTusResumeState(sidecar)
			uploadURL = ""
			restartedAfterComplete = true
			continue
		}

		// (4) Stream the remaining bytes to the vault via PATCH, reading directly
		// from disk so the recording is never fully buffered in memory. When a chunk
		// size is configured the data is sent across several PATCH requests,
		// checkpointing the offset after each one so an interruption resumes from the
		// last completed chunk instead of re-uploading everything.
		chunkSize := tusChunkSize()
		patchFailed := false
		var lastBody string
		loggedProgressBucket := tusProgressBucket(offset, size)
		for offset < size {
			// Re-seek every chunk so the on-disk position always matches the
			// server-acknowledged offset, even if a PATCH was partially accepted.
			if _, sErr := file.Seek(offset, io.SeekStart); sErr != nil {
				return false, false, true, "", sErr
			}
			patchLen := size - offset
			if chunkSize > 0 && chunkSize < patchLen {
				patchLen = chunkSize
			}
			newOffset, status, respBody, perr := tusPatch(client, uploadURL, offset, patchLen, file, setHeaders)
			lastStatus = status
			if perr != nil {
				if status >= 400 {
					// Definitive rejection (e.g. provider push failed during finalize).
					// Re-evaluate via HEAD on the next iteration to decide retry/restart.
					log.Log.Info(label + ": resumable patch rejected, " + perr.Error())
				} else {
					log.Log.Info(label + ": resumable patch failed, " + perr.Error())
				}
				tusBackoff(attempt)
				patchFailed = true
				break
			}
			offset = newOffset
			if offset > highWaterOffset {
				highWaterOffset = offset
			}
			lastBody = respBody
			logTusUploadProgress(label, offset, size, &loggedProgressBucket)
			if offset < size {
				// Partial progress: persist so a later retry resumes from here.
				saveTusResumeState(sidecar, tusResumeState{UploadURL: uploadURL, VaultURI: baseURL, Size: size})
			}
		}
		if patchFailed {
			if highWaterOffset > startHighWater {
				// Genuine net progress (we advanced past the furthest point any previous
				// attempt reached) refreshes the retry budget: maxAttempts bounds the
				// number of consecutive *non-progressing* failures, not the number of
				// chunks needed for a large recording. A server that keeps rejecting the
				// same offset (no net progress, e.g. a persistent ERR_MISMATCHED_OFFSET)
				// therefore gives up after maxAttempts instead of retrying forever.
				attempt = -1
			}
			continue
		}

		// All declared bytes have been sent and acknowledged: the upload is done.
		removeTusResumeState(sidecar)
		return true, true, true, lastBody, nil
	}

	// Every attempt failed. Only report responded=true when the vault actually
	// answered on the last attempt (lastStatus > 0). If every attempt failed at
	// the transport level (lastStatus == 0, e.g. the internet was disconnected),
	// report responded=false so the caller keeps the recording queued and retries
	// later instead of consuming its retry budget and entering the long back-off
	// timeout.
	return false, lastStatus > 0, true, "resumable upload did not complete after retries", errors.New(label + ": resumable upload did not complete after retries")
}

// uploadVaultResumable uploads a recording directly to a Kerberos Vault using
// the tus resumable upload protocol. Credentials travel in the
// X-Kerberos-Storage-* headers on every request and routing (directory/provider)
// is additionally carried in the tus Upload-Metadata.
func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName, label, slot string) (bool, bool, bool, string, error) {
	baseURL := strings.TrimRight(vault.URI, "/") + tusUploadPath
	metadataValues := map[string]string{
		"filename":  fileName,
		"device":    deviceKey,
		"directory": vault.Directory,
		"provider":  vault.Provider,
		"capture":   "IPCamera",
		"cloudkey":  publicKey,
		"fps":       queuedRecordingFPS(fileName),
	}
	addRecordingTusMetadata(metadataValues, fileName)
	metadata := encodeTusMetadata(metadataValues)
	setHeaders := func(h http.Header, fn string) {
		setVaultTusHeaders(h, vault, publicKey, deviceKey, fn)
	}
	return runTusUpload(baseURL, metadata, fileName, label, slot, setHeaders)
}

// uploadHubResumable uploads a recording to Kerberos Hub's tus endpoint, which
// authenticates the agent with its Hub public/private key and proxies the
// resumable upload to the Kerberos Vault on the agent's behalf. The vault
// directory and provider are resolved and injected by Kerberos Hub, so they are
// intentionally omitted from the metadata here.
func uploadHubResumable(config *models.Config, fileName, label, slot string) (bool, bool, bool, string, error) {
	baseURL := strings.TrimRight(config.HubURI, "/") + tusUploadPath
	metadataValues := map[string]string{
		"filename": fileName,
		"device":   config.Key,
		"capture":  "IPCamera",
		"fps":      queuedRecordingFPS(fileName),
	}
	addRecordingTusMetadata(metadataValues, fileName)
	metadata := encodeTusMetadata(metadataValues)
	setHeaders := func(h http.Header, fn string) {
		setHubTusHeaders(h, config, fn)
	}
	return runTusUpload(baseURL, metadata, fileName, label, slot, setHeaders)
}

func addRecordingTusMetadata(values map[string]string, fileName string) {
	metadata, ok := queuedRecordingMetadata(fileName)
	if !ok {
		return
	}
	if metadata.Duration > 0 {
		values["duration"] = strconv.FormatUint(metadata.Duration, 10)
	}
	if metadata.Timestamp > 0 {
		values["timestamp"] = strconv.FormatInt(metadata.Timestamp, 10)
	}
}

// tusCreate performs the tus "creation" request (POST). On success it returns
// the resolved upload URL the agent should use for subsequent HEAD/PATCH calls.
func tusCreate(client *http.Client, baseURL string, size int64, metadata string, setHeaders tusHeaderFunc, fileName string) (string, int, error) {
	req, err := http.NewRequest("POST", baseURL, nil)
	if err != nil {
		return "", 0, err
	}
	req.Header.Set("Tus-Resumable", tusResumableVersion)
	req.Header.Set("Upload-Length", strconv.FormatInt(size, 10))
	if metadata != "" {
		req.Header.Set("Upload-Metadata", metadata)
	}
	setHeaders(req.Header, fileName)

	resp, err := client.Do(req)
	if resp != nil {
		defer resp.Body.Close()
	}
	if err != nil {
		return "", 0, err
	}
	io.Copy(io.Discard, resp.Body)

	if resp.StatusCode != http.StatusCreated {
		return "", resp.StatusCode, fmt.Errorf("unexpected status creating upload: %s", resp.Status)
	}
	location := resp.Header.Get("Location")
	if location == "" {
		return "", resp.StatusCode, errors.New("missing Location header in create response")
	}
	return resolveTusLocation(baseURL, location), resp.StatusCode, nil
}

// tusHead performs the tus "offset" request (HEAD) and returns the current
// server-side upload offset.
func tusHead(client *http.Client, uploadURL string, setHeaders tusHeaderFunc) (int64, int, error) {
	req, err := http.NewRequest("HEAD", uploadURL, nil)
	if err != nil {
		return 0, 0, err
	}
	req.Header.Set("Tus-Resumable", tusResumableVersion)
	setHeaders(req.Header, "")

	resp, err := client.Do(req)
	if resp != nil {
		defer resp.Body.Close()
	}
	if err != nil {
		return 0, 0, err
	}
	io.Copy(io.Discard, resp.Body)

	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent {
		return 0, resp.StatusCode, fmt.Errorf("unexpected status on HEAD: %s", resp.Status)
	}
	offsetStr := resp.Header.Get("Upload-Offset")
	offset, perr := strconv.ParseInt(offsetStr, 10, 64)
	if perr != nil {
		return 0, resp.StatusCode, fmt.Errorf("invalid Upload-Offset header: %q", offsetStr)
	}
	return offset, resp.StatusCode, nil
}

// tusPatch streams up to length bytes of the file (starting at offset) to the
// upload URL using a single PATCH request. The body is read straight from the
// *os.File, so the recording is never fully buffered in memory.
func tusPatch(client *http.Client, uploadURL string, offset, length int64, file io.Reader, setHeaders tusHeaderFunc) (int64, int, string, error) {
	req, err := http.NewRequest("PATCH", uploadURL, io.LimitReader(file, length))
	if err != nil {
		return offset, 0, "", err
	}
	req.ContentLength = length
	req.Header.Set("Tus-Resumable", tusResumableVersion)
	req.Header.Set("Content-Type", "application/offset+octet-stream")
	req.Header.Set("Upload-Offset", strconv.FormatInt(offset, 10))
	setHeaders(req.Header, "")

	resp, err := client.Do(req)
	if resp != nil {
		defer resp.Body.Close()
	}
	if err != nil {
		return offset, 0, "", err
	}
	bodyBytes, _ := io.ReadAll(resp.Body)
	respBody := string(bodyBytes)

	if resp.StatusCode != http.StatusNoContent {
		return offset, resp.StatusCode, respBody, fmt.Errorf("unexpected status on PATCH: %s, %s", resp.Status, respBody)
	}
	newOffsetStr := resp.Header.Get("Upload-Offset")
	newOffset, perr := strconv.ParseInt(newOffsetStr, 10, 64)
	if perr != nil {
		// A 204 without a parseable offset means this PATCH was fully accepted.
		return offset + length, resp.StatusCode, respBody, nil
	}
	return newOffset, resp.StatusCode, respBody, nil
}

// tusTerminate best-effort deletes an upload server-side (DELETE).
func tusTerminate(client *http.Client, uploadURL string, setHeaders tusHeaderFunc) {
	req, err := http.NewRequest("DELETE", uploadURL, nil)
	if err != nil {
		return
	}
	req.Header.Set("Tus-Resumable", tusResumableVersion)
	setHeaders(req.Header, "")

	resp, derr := client.Do(req)
	if resp != nil {
		io.Copy(io.Discard, resp.Body)
		resp.Body.Close()
	}
	_ = derr
}

// setVaultTusHeaders sets the Kerberos Vault authentication and routing headers
// on every tus request. Credentials are sent on each request (and never stored
// server-side in the upload metadata). When fileName is empty it is omitted, as
// it is only useful on the creation request (routing also travels in the tus
// Upload-Metadata).
func setVaultTusHeaders(h http.Header, vault models.KStorage, publicKey, deviceKey, fileName string) {
	h.Set("X-Kerberos-Storage-CloudKey", publicKey)
	h.Set("X-Kerberos-Storage-AccessKey", vault.AccessKey)
	h.Set("X-Kerberos-Storage-SecretAccessKey", vault.SecretAccessKey)
	h.Set("X-Kerberos-Storage-Provider", vault.Provider)
	h.Set("X-Kerberos-Storage-Device", deviceKey)
	h.Set("X-Kerberos-Storage-Directory", vault.Directory)
	h.Set("X-Kerberos-Storage-Capture", "IPCamera")
	if fileName != "" {
		h.Set("X-Kerberos-Storage-FileName", fileName)
	}
}

// setHubTusHeaders sets the Kerberos Hub authentication headers on every tus
// request of a hub-proxied resumable upload. The agent authenticates with its
// Hub public/private key (exactly as the legacy single-POST hub upload does);
// Kerberos Hub validates the subscription and injects the vault credentials and
// directory/provider on the agent's behalf.
func setHubTusHeaders(h http.Header, config *models.Config, fileName string) {
	h.Set("X-Kerberos-Hub-PublicKey", config.HubKey)
	h.Set("X-Kerberos-Hub-PrivateKey", config.HubPrivateKey)
	h.Set("X-Kerberos-Hub-Region", config.S3.Region)
	h.Set("X-Kerberos-Storage-Device", config.Key)
	h.Set("X-Kerberos-Storage-Capture", "IPCamera")
	if fileName != "" {
		h.Set("X-Kerberos-Storage-FileName", fileName)
	}
}

// encodeTusMetadata serializes a map into the tus Upload-Metadata header format:
// a comma separated list of "key base64(value)" pairs. Keys are sorted for a
// deterministic header value. Empty values are skipped.
func encodeTusMetadata(pairs map[string]string) string {
	parts := make([]string, 0, len(pairs))
	for k, v := range pairs {
		if v == "" {
			continue
		}
		parts = append(parts, k+" "+base64.StdEncoding.EncodeToString([]byte(v)))
	}
	sort.Strings(parts)
	return strings.Join(parts, ",")
}

// resolveTusLocation turns the Location header returned by the create request
// into an absolute URL. To keep talking to the agent's configured vault host
// (and avoid issues when the vault sits behind a proxy that rewrites the host),
// it keeps the configured base URL and only appends the server-assigned upload
// id taken from the Location.
func resolveTusLocation(baseURL, location string) string {
	if ref, err := url.Parse(location); err == nil {
		trimmed := strings.Trim(ref.Path, "/")
		if trimmed != "" {
			segments := strings.Split(trimmed, "/")
			id := segments[len(segments)-1]
			if id != "" {
				return strings.TrimRight(baseURL, "/") + "/" + id
			}
		}
	}
	// Fallback: resolve the reference against the base URL as-is.
	if base, err := url.Parse(baseURL); err == nil {
		if ref, err := url.Parse(location); err == nil {
			return base.ResolveReference(ref).String()
		}
	}
	return location
}

// tusSidecarDir is the directory where resume state files are kept. It is
// intentionally separate from data/cloud (which is scanned for recordings to
// upload) so the sidecar files are never mistaken for recordings.
func tusSidecarDir() string {
	return "data/tus"
}

func tusSidecarPath(fileName, slot string) string {
	safe := strings.ReplaceAll(fileName, "/", "_")
	safe = strings.ReplaceAll(safe, string(os.PathSeparator), "_")
	return filepath.Join(tusSidecarDir(), safe+"."+slot+".json")
}

// loadTusResumeState returns a previously stored upload URL for the given
// sidecar, but only if it was created against the same vault base URL. Any
// mismatch or read/parse error yields an empty string (start fresh).
func loadTusResumeState(path, baseURL string) string {
	b, err := os.ReadFile(path)
	if err != nil {
		return ""
	}
	var state tusResumeState
	if err := json.Unmarshal(b, &state); err != nil {
		return ""
	}
	if state.UploadURL == "" || state.VaultURI != baseURL {
		return ""
	}
	return state.UploadURL
}

func saveTusResumeState(path string, state tusResumeState) {
	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
		return
	}
	b, err := json.Marshal(state)
	if err != nil {
		return
	}
	_ = os.WriteFile(path, b, 0o644)
}

func removeTusResumeState(path string) {
	_ = os.Remove(path)
}

// tusBackoffBaseDelay is the base delay used by tusBackoff for the exponential
// back-off between resume attempts. It is a package variable (rather than a
// constant) so tests can shrink it to keep them fast.
var tusBackoffBaseDelay = 500 * time.Millisecond

// tusBackoff sleeps for an exponentially increasing duration (capped) between
// resume attempts to avoid hammering a temporarily unavailable vault.
func tusBackoff(attempt int) {
	delay := tusBackoffBaseDelay * time.Duration(1<<uint(attempt))
	if delay > 3*time.Second {
		delay = 3 * time.Second
	}
	time.Sleep(delay)
}
