beam/main.go

package main

import (
	"context"
	"crypto/rand"
	_ "embed"
	"encoding/json"
	"errors"
	"fmt"
	"html/template"
	"io"
	"io/fs"
	"log"
	"mime"
	"mime/multipart"
	"net/http"
	"net/url"
	"os"
	"os/signal"
	"path/filepath"
	"strconv"
	"strings"
	"syscall"
	"time"
)

var uploadDir string

const idLength = 8
const idCharset = "abcdefghijklmnopqrstuvwxyz0123456789"

// defaultMaxUploadSize is used when MAX_UPLOAD_SIZE is not set.
const defaultMaxUploadSize int64 = 5 << 30 // 5GB

// maxMultipartMemory caps how much of a multipart upload is buffered in
// memory before spilling to temporary files on disk, regardless of
// maxUploadSize.
const maxMultipartMemory int64 = 32 << 20 // 32MB

// maxFilesPerUpload bounds how many parts a single multipart upload may
// contain, so one request cannot exhaust inodes with tiny files.
const maxFilesPerUpload = 100

// defaultShutdownGrace is used when SHUTDOWN_GRACE is not set. Align it with
// the pod's terminationGracePeriodSeconds when uploads are expected to be
// large and slow.
const defaultShutdownGrace = 15 * time.Second

// defaultCleanupInterval is used when CLEANUP_INTERVAL is not set.
const defaultCleanupInterval = 6 * time.Hour

// defaultFileExpires is used when FILE_DURATION is not set.
const defaultFileExpires = 7 * 24 * time.Hour

// newUploadGrace keeps the cleanup sweep from removing an upload directory
// that was just created but whose file has not been written yet.
const newUploadGrace = 5 * time.Minute

// Embedded so the binary does not depend on its working directory.
//
//go:embed index.html.tmpl
var indexTemplate string

var indexHTML = template.Must(template.New("index").Parse(indexTemplate))
var maxUploadSize int64 = defaultMaxUploadSize
var fileExpires = defaultFileExpires
var version = "dev"

func main() {
	uploadDir = os.Getenv("UPLOAD_DIR")
	if uploadDir == "" {
		uploadDir = "/uploads"
	}
	// DRYRUN_CLEAN defaults to true: cleanup reports what it would delete but
	// removes nothing until it is explicitly disabled.
	dryRun := true
	if dryRunEnv := os.Getenv("DRYRUN_CLEAN"); dryRunEnv != "" {
		parsed, err := strconv.ParseBool(dryRunEnv)
		if err != nil {
			log.Fatalf("DRYRUN_CLEAN %s is invalid: %v", dryRunEnv, err)
		}
		dryRun = parsed
	}

	fileDuration := os.Getenv("FILE_DURATION")
	if fileDuration != "" {
		expires, err := time.ParseDuration(fileDuration)
		if err != nil {
			log.Fatalf("FILE_DURATION %s is invalid: %v", fileDuration, err)
		}
		// A negative duration would make every file look expired on the very
		// first sweep and wipe the upload directory.
		if expires < 0 {
			log.Fatalf("FILE_DURATION %s is invalid: must not be negative", fileDuration)
		}
		fileExpires = expires
	}

	if maxUploadSizeEnv := os.Getenv("MAX_UPLOAD_SIZE"); maxUploadSizeEnv != "" {
		size, err := strconv.ParseInt(maxUploadSizeEnv, 10, 64)
		if err != nil {
			log.Fatalf("MAX_UPLOAD_SIZE %s is invalid: %v", maxUploadSizeEnv, err)
		}
		if size <= 0 {
			log.Fatalf("MAX_UPLOAD_SIZE %s is invalid: must be positive", maxUploadSizeEnv)
		}
		maxUploadSize = size
	}

	shutdownGrace := defaultShutdownGrace
	if graceEnv := os.Getenv("SHUTDOWN_GRACE"); graceEnv != "" {
		grace, err := time.ParseDuration(graceEnv)
		if err != nil {
			log.Fatalf("SHUTDOWN_GRACE %s is invalid: %v", graceEnv, err)
		}
		if grace <= 0 {
			log.Fatalf("SHUTDOWN_GRACE %s is invalid: must be positive", graceEnv)
		}
		shutdownGrace = grace
	}

	cleanupInterval := defaultCleanupInterval
	if intervalEnv := os.Getenv("CLEANUP_INTERVAL"); intervalEnv != "" {
		interval, err := time.ParseDuration(intervalEnv)
		if err != nil {
			log.Fatalf("CLEANUP_INTERVAL %s is invalid: %v", intervalEnv, err)
		}
		if interval <= 0 {
			log.Fatalf("CLEANUP_INTERVAL %s is invalid: must be positive", intervalEnv)
		}
		cleanupInterval = interval
	}

	port := os.Getenv("PORT")
	if port == "" {
		port = "8080"
	}

	if err := os.MkdirAll(uploadDir, 0755); err != nil {
		log.Fatalf("Failed to create upload directory: %v", err)
	}
	mux := http.NewServeMux()
	mux.HandleFunc("/", handleRequest)

	log.Printf("Starting server on port %s, upload directory: %s", port, uploadDir)
	log.Printf("Version: %s", version)
	log.Printf("Files expires after: %v (DRYRUN_CLEAN=%v)", fileExpires, dryRun)
	log.Printf("Max upload size: %d bytes", maxUploadSize)

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	if fileExpires != 0 {
		log.Printf("Cleanup interval: %v", cleanupInterval)
		go cleanupProcedure(ctx, fileExpires, cleanupInterval, dryRun)
	} else {
		log.Print("Cleaning old files is disabled")
	}

	// Note: ReadTimeout/WriteTimeout are intentionally not set since they
	// would cap the total duration of large file uploads/downloads.
	// ReadHeaderTimeout and IdleTimeout still protect against slow-loris
	// style attacks on connection/header handling.
	srv := &http.Server{
		Addr:              ":" + port,
		Handler:           withLogging(mux),
		ReadHeaderTimeout: 10 * time.Second,
		IdleTimeout:       120 * time.Second,
	}

	serverErr := make(chan error, 1)
	go func() {
		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			serverErr <- err
		}
		close(serverErr)
	}()

	select {
	case err := <-serverErr:
		log.Fatalf("Server failed: %v", err)
	case <-ctx.Done():
		log.Print("Shutdown signal received, shutting down gracefully...")
		shutdownCtx, cancel := context.WithTimeout(context.Background(), shutdownGrace)
		defer cancel()
		// A large upload in flight will not finish inside the grace period.
		// Force the remaining connections closed and still exit cleanly, so a
		// rollout mid-transfer does not look like a crash to Kubernetes.
		if err := srv.Shutdown(shutdownCtx); err != nil {
			log.Printf("Graceful shutdown did not complete within %v (%v), forcing close", shutdownGrace, err)
			srv.Close()
		}
		log.Print("Server stopped")
	}
}

func handleRequest(w http.ResponseWriter, r *http.Request) {
	if r.Method == http.MethodGet && r.URL.Path == "/" {
		serveFrontpage(w, r)
		return
	}
	if r.Method == http.MethodGet && r.URL.Path == "/healthz" {
		handleHealthz(w, r)
		return
	}

	if r.Method == http.MethodPost && r.URL.Path == "/upload" {
		handleMultipartUpload(w, r)
		return
	}

	if r.Method == http.MethodPut {
		handleUpload(w, r)
		return
	}

	// ServeContent handles HEAD correctly, so downloads answer both verbs.
	if r.Method == http.MethodGet || r.Method == http.MethodHead {
		handleDownload(w, r)
		return
	}

	w.Header().Set("Allow", "GET, HEAD, PUT, POST")
	http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
}

func handleHealthz(w http.ResponseWriter, r *http.Request) {
	if _, err := os.Stat(uploadDir); err != nil {
		log.Printf("Healthz check failed: %v", err)
		http.Error(w, "Upload directory is not accessible", http.StatusServiceUnavailable)
		return
	}
	w.WriteHeader(http.StatusOK)
}

func handleUpload(w http.ResponseWriter, r *http.Request) {
	rawName := strings.TrimPrefix(r.URL.Path, "/")
	if rawName == "" {
		http.Error(w, "Filename required", http.StatusBadRequest)
		return
	}

	filename, ok := safeFilename(rawName)
	if !ok {
		http.Error(w, "Invalid filename", http.StatusBadRequest)
		return
	}

	id, err := generateUniqueID()
	if err != nil {
		log.Printf("Failed to generate unique ID: %v", err)
		http.Error(w, "Failed to generate unique ID", http.StatusInternalServerError)
		return
	}

	targetDir := filepath.Join(uploadDir, id)
	targetPath := filepath.Join(targetDir, filename)
	file, err := os.Create(targetPath)
	if err != nil {
		os.Remove(targetDir)
		log.Printf("Failed to create %q: %v", targetPath, err)
		http.Error(w, "Failed to create file", http.StatusInternalServerError)
		return
	}

	// MaxBytesReader needs the ResponseWriter the stdlib handed us so it can
	// mark the connection for close; the logging wrapper would hide it.
	limitedBody := http.MaxBytesReader(baseWriter(w), r.Body, maxUploadSize)
	written, err := io.Copy(file, limitedBody)
	// Closed here rather than in a defer: nothing above returns early, and a
	// write-back failure only surfacing at close would otherwise arrive after
	// the 200 and the URL had gone out, leaving the client believing a
	// truncated file was stored.
	closeErr := file.Close()
	if err == nil {
		err = closeErr
	}
	if err != nil {
		os.Remove(targetPath)
		os.Remove(targetDir)
		if isMaxBytesError(err) {
			http.Error(w, fmt.Sprintf("File exceeds maximum upload size of %d bytes", maxUploadSize), http.StatusRequestEntityTooLarge)
			return
		}
		log.Printf("Failed to write %q: %v", targetPath, err)
		http.Error(w, "Failed to write file", http.StatusInternalServerError)
		return
	}

	link := downloadURL(r, id, filename)

	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
	w.WriteHeader(http.StatusOK)
	w.Write([]byte(link + "\n"))

	log.Printf("Uploaded %q (%d bytes) -> %s", filename, written, link)
}

// safeFilename reduces an untrusted name to its base component and reports
// whether the result is safe to join onto the upload directory. Note that
// filepath.IsLocal alone is not enough: it accepts ".", and filepath.Base
// turns an empty name into "." as well.
func safeFilename(name string) (string, bool) {
	base := filepath.Base(name)
	if base == "." || base == ".." || !filepath.IsLocal(base) {
		return "", false
	}
	return base, true
}

// downloadURL builds the externally visible URL for a stored file. The
// filename is percent-escaped so that spaces, '#' and '?' survive the round
// trip back through handleDownload.
func downloadURL(r *http.Request, id, filename string) string {
	return fmt.Sprintf("%s://%s/%s/%s", getProtocol(r), getHost(r), id, url.PathEscape(filename))
}

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

// uploadResponse is the success body for POST /upload. Failed carries the
// names of parts that could not be stored, so a partial failure is visible to
// the client instead of silently yielding a short urls list.
type uploadResponse struct {
	URLs   []string `json:"urls"`
	Failed []string `json:"failed,omitempty"`
}

func handleMultipartUpload(w http.ResponseWriter, r *http.Request) {
	// See handleUpload: MaxBytesReader must see the original ResponseWriter.
	r.Body = http.MaxBytesReader(baseWriter(w), r.Body, maxUploadSize)

	memoryLimit := maxUploadSize
	if memoryLimit > maxMultipartMemory {
		memoryLimit = maxMultipartMemory
	}

	// Registered before ParseMultipartForm: a parse failure can still leave
	// spilled temporary files behind, and the error paths below return early.
	defer func() {
		if r.MultipartForm != nil {
			r.MultipartForm.RemoveAll()
		}
	}()

	if err := r.ParseMultipartForm(memoryLimit); err != nil {
		if isMaxBytesError(err) {
			writeMultipartError(w, http.StatusRequestEntityTooLarge, fmt.Sprintf("Upload exceeds maximum size of %d bytes", maxUploadSize))
			return
		}
		writeMultipartError(w, http.StatusBadRequest, "Failed to parse multipart form")
		return
	}

	files := r.MultipartForm.File["files"]
	if len(files) == 0 {
		writeMultipartError(w, http.StatusBadRequest, "No files provided")
		return
	}
	if len(files) > maxFilesPerUpload {
		writeMultipartError(w, http.StatusBadRequest, fmt.Sprintf("Too many files: %d, maximum is %d", len(files), maxFilesPerUpload))
		return
	}

	resp := uploadResponse{URLs: make([]string, 0, len(files))}

	for _, fileHeader := range files {
		filename, ok := safeFilename(fileHeader.Filename)
		if !ok {
			log.Printf("Rejected upload part with invalid filename %q", fileHeader.Filename)
			resp.Failed = append(resp.Failed, fileHeader.Filename)
			continue
		}

		link, err := storeUploadedFile(r, fileHeader, filename)
		if err != nil {
			log.Printf("Failed to store uploaded file %q: %v", filename, err)
			resp.Failed = append(resp.Failed, filename)
			continue
		}
		resp.URLs = append(resp.URLs, link)
	}

	// Every part failed: this is a server-side problem, not a 200.
	if len(resp.URLs) == 0 {
		writeMultipartError(w, http.StatusInternalServerError, "Failed to store uploaded files")
		return
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(resp)
}

// storeUploadedFile writes one multipart part to its own upload directory and
// returns the download URL. Any partially written file or directory is removed
// before returning an error.
func storeUploadedFile(r *http.Request, fileHeader *multipart.FileHeader, filename string) (string, error) {
	src, err := fileHeader.Open()
	if err != nil {
		return "", err
	}
	defer src.Close()

	id, err := generateUniqueID()
	if err != nil {
		return "", err
	}

	targetDir := filepath.Join(uploadDir, id)
	targetPath := filepath.Join(targetDir, filename)

	dst, err := os.Create(targetPath)
	if err != nil {
		os.Remove(targetDir)
		return "", err
	}

	written, err := io.Copy(dst, src)
	closeErr := dst.Close()
	if err == nil {
		err = closeErr
	}
	if err != nil {
		os.Remove(targetPath)
		os.Remove(targetDir)
		return "", err
	}

	link := downloadURL(r, id, filename)
	log.Printf("Uploaded %q (%d bytes) -> %s", filename, written, link)
	return link, nil
}

func handleDownload(w http.ResponseWriter, r *http.Request) {
	parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/"), "/")
	if len(parts) != 2 {
		http.NotFound(w, r)
		return
	}

	id, idOK := safeFilename(parts[0])
	filename, nameOK := safeFilename(parts[1])
	if !idOK || !nameOK {
		http.NotFound(w, r)
		return
	}

	filePath := filepath.Join(uploadDir, id, filename)
	file, err := os.Open(filePath)
	if err != nil {
		http.NotFound(w, r)
		return
	}
	defer file.Close()

	stat, err := file.Stat()
	if err != nil {
		log.Printf("Failed to stat %q: %v", filePath, err)
		http.Error(w, "Failed to stat file", http.StatusInternalServerError)
		return
	}
	// A directory would otherwise be served as a zero-length file.
	if stat.IsDir() {
		http.NotFound(w, r)
		return
	}

	// Set Content-Type explicitly so ServeContent does not sniff the body.
	// The extension lookup relies on the system MIME database, which the
	// Dockerfile provides via the mailcap package.
	w.Header().Set("Content-Type", detectContentType(filename))
	// Uploads are anonymous and served back from this origin, so an uploaded
	// .html or .svg opened inline would otherwise be stored XSS. The sandbox
	// directive puts the response in an opaque origin with scripting, forms and
	// top-level navigation disabled; nosniff stops a mislabeled file from being
	// re-typed into something executable. Both are set before ServeContent so
	// they apply to its 206, 304 and HEAD responses too.
	w.Header().Set("X-Content-Type-Options", "nosniff")
	w.Header().Set("Content-Security-Policy", "sandbox")
	// FormatMediaType returns "" for names it cannot encode; don't emit an
	// empty header in that case.
	if disposition := mime.FormatMediaType("inline", map[string]string{"filename": filename}); disposition != "" {
		w.Header().Set("Content-Disposition", disposition)
	}
	// Download URLs are unguessable secrets, so keep shared caches out of it.
	w.Header().Set("Cache-Control", "private")

	// ServeContent handles Range, If-Modified-Since, Content-Length and HEAD,
	// which makes interrupted downloads resumable with `curl -C -`.
	http.ServeContent(w, r, filename, stat.ModTime(), file)
}

func generateUniqueID() (string, error) {
	const maxRetries = 100

	for range maxRetries {
		id := generateID()
		targetDir := filepath.Join(uploadDir, id)

		if err := os.Mkdir(targetDir, 0755); err == nil {
			return id, nil
		} else if !os.IsExist(err) {
			return "", err
		}
	}

	return "", fmt.Errorf("failed to generate unique ID after %d retries", maxRetries)
}

func generateID() string {
	bytes := make([]byte, idLength)
	if _, err := rand.Read(bytes); err != nil {
		panic(err)
	}

	for i, b := range bytes {
		bytes[i] = idCharset[int(b)%len(idCharset)]
	}

	return string(bytes)
}

func getHost(r *http.Request) string {
	// Check X-Forwarded-Host (common proxy header)
	if host := r.Header.Get("X-Forwarded-Host"); host != "" {
		return host
	}

	// Check standard Forwarded header (RFC 7239)
	if forwarded := r.Header.Get("Forwarded"); forwarded != "" {
		// Parse	 "Forwarded: for=...; host=example.com; proto=https"
		parts := strings.SplitSeq(forwarded, ";")
		for part := range parts {
			part = strings.TrimSpace(part)
			if host, found := strings.CutPrefix(part, "host="); found {
				return host
			}
		}
	}

	// Fallback to Host header
	if host := r.Host; host != "" {
		return host
	}

	return "localhost:8080"
}

func getProtocol(r *http.Request) string {
	if proto := r.Header.Get("X-Forwarded-Proto"); proto != "" {
		return proto
	}

	if forwarded := r.Header.Get("Forwarded"); forwarded != "" {
		parts := strings.SplitSeq(forwarded, ";")
		for part := range parts {
			part = strings.TrimSpace(part)
			if proto, found := strings.CutPrefix(part, "proto="); found {
				return proto
			}
		}
	}

	if ssl := r.Header.Get("X-Forwarded-SSL"); ssl == "on" {
		return "https"
	}
	if r.TLS != nil {
		return "https"
	}

	return "http"
}

// isMaxBytesError reports whether err was caused by exceeding the limit
// set on an http.MaxBytesReader.
func isMaxBytesError(err error) bool {
	var maxBytesErr *http.MaxBytesError
	return errors.As(err, &maxBytesErr)
}

func detectContentType(filename string) string {
	ext := filepath.Ext(filename)
	mimeType := mime.TypeByExtension(ext)
	if mimeType != "" {
		return mimeType
	}
	return "application/octet-stream"
}

// responseRecorder captures the status code and byte count actually written,
// so the access log reports what happened rather than what the handler
// intended. http.ServeContent in particular may answer 206 or 304.
type responseRecorder struct {
	http.ResponseWriter
	status  int
	written int64
}

func (rr *responseRecorder) WriteHeader(status int) {
	if rr.status == 0 {
		rr.status = status
	}
	rr.ResponseWriter.WriteHeader(status)
}

func (rr *responseRecorder) Write(b []byte) (int, error) {
	if rr.status == 0 {
		rr.status = http.StatusOK
	}
	n, err := rr.ResponseWriter.Write(b)
	rr.written += int64(n)
	return n, err
}

// Unwrap lets http.ResponseController and baseWriter reach the original
// ResponseWriter.
func (rr *responseRecorder) Unwrap() http.ResponseWriter { return rr.ResponseWriter }

// baseWriter returns the ResponseWriter the net/http server created. It is
// needed by http.MaxBytesReader, which type-asserts its argument to an
// unexported interface to mark an oversized request's connection for close;
// that assertion fails against a wrapper.
func baseWriter(w http.ResponseWriter) http.ResponseWriter {
	for {
		u, ok := w.(interface{ Unwrap() http.ResponseWriter })
		if !ok {
			return w
		}
		w = u.Unwrap()
	}
}

// withLogging records one access-log line per request, including for handlers
// that never write an explicit status.
func withLogging(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		start := time.Now()
		rec := &responseRecorder{ResponseWriter: w}

		next.ServeHTTP(rec, r)

		status := rec.status
		if status == 0 {
			status = http.StatusOK
		}
		// %q on the path: it is attacker-controlled and would otherwise let a
		// request forge log lines by embedding newlines.
		log.Printf("%s %q %d %d %v", r.Method, r.URL.Path, status, rec.written, time.Since(start))
	})
}

func serveFrontpage(w http.ResponseWriter, r *http.Request) {
	pageData := map[string]any{
		"Proto":         getProtocol(r),
		"Host":          getHost(r),
		"MaxUploadSize": maxUploadSize,
		"Expiry":        formatExpiry(fileExpires),
		"Version":       version,
	}

	w.Header().Set("Content-Type", "text/html; charset=utf-8")
	if err := indexHTML.Execute(w, pageData); err != nil {
		log.Printf("Failed to render front page: %v", err)
	}
}

// formatExpiry renders a retention duration for the front page.
// Duration.String() would produce "168h0m0s" for the default week.
func formatExpiry(d time.Duration) string {
	switch {
	case d <= 0:
		return "never"
	case d%(24*time.Hour) == 0:
		return plural(int(d/(24*time.Hour)), "day")
	case d%time.Hour == 0:
		return plural(int(d/time.Hour), "hour")
	case d%time.Minute == 0:
		return plural(int(d/time.Minute), "minute")
	default:
		return d.String()
	}
}

func plural(n int, unit string) string {
	if n == 1 {
		return "1 " + unit
	}
	return fmt.Sprintf("%d %ss", n, unit)
}

// Cleanup

func isDirEmpty(name string) (bool, error) {
	f, err := os.Open(name)
	if err != nil {
		return false, err
	}
	defer f.Close()

	// Read exactly one entry.
	_, err = f.ReadDir(1)
	if errors.Is(err, io.EOF) {
		// EOF means there are no entries, so the directory is empty.
		return true, nil
	}

	// If err is nil, an entry was found (not empty).
	// If err is something else, a real error occurred.
	return false, err
}

func cleanupProcedure(ctx context.Context, duration, interval time.Duration, dryRun bool) {
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		cleanupCycle(duration, dryRun)

		select {
		case <-ctx.Done():
			log.Print("Cleanup stopped")
			return
		case <-ticker.C:
		}
	}
}

// cleanupCycle performs one sweep: expired files first, then the upload
// directories left empty behind them.
func cleanupCycle(duration time.Duration, dryRun bool) {
	oldFiles, err := collectExpiredFiles(uploadDir, duration)
	if err != nil {
		log.Printf("Cleanup: failed to scan %q for expired files: %v", uploadDir, err)
	}

	removedFiles := removeAll(oldFiles, "expired file", dryRun)

	emptyDirs, err := collectEmptyDirs(uploadDir)
	if err != nil {
		log.Printf("Cleanup: failed to scan %q for empty folders: %v", uploadDir, err)
	}

	removedDirs := removeAll(emptyDirs, "empty folder", dryRun)

	log.Printf("Cleanup finished: %d expired file(s) and %d empty folder(s) found, %d and %d removed (dryRun=%v)",
		len(oldFiles), len(emptyDirs), removedFiles, removedDirs, dryRun)
}

// collectExpiredFiles returns files under root whose mtime is at least
// duration old.
func collectExpiredFiles(root string, duration time.Duration) ([]string, error) {
	var oldFiles []string

	err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
		if err != nil {
			log.Printf("Cleanup: skipping %q: %v", path, err)
			return nil
		}
		if d.IsDir() {
			return nil
		}

		info, err := d.Info()
		if err != nil {
			log.Printf("Cleanup: failed to stat %q: %v", path, err)
			return nil
		}
		if time.Since(info.ModTime()) >= duration {
			oldFiles = append(oldFiles, path)
		}
		return nil
	})

	return oldFiles, err
}

// collectEmptyDirs returns empty directories under root, excluding root itself
// and any directory created within newUploadGrace, which may belong to an
// upload that has not written its file yet.
func collectEmptyDirs(root string) ([]string, error) {
	var emptyDirs []string

	err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
		// d is nil when the root itself cannot be read, so this check must
		// come first.
		if err != nil {
			log.Printf("Cleanup: skipping %q: %v", path, err)
			return nil
		}
		if !d.IsDir() {
			return nil
		}
		if filepath.Clean(path) == filepath.Clean(root) {
			return nil
		}

		info, err := d.Info()
		if err != nil {
			log.Printf("Cleanup: failed to stat %q: %v", path, err)
			return nil
		}
		if time.Since(info.ModTime()) < newUploadGrace {
			return nil
		}

		empty, err := isDirEmpty(path)
		if err != nil {
			log.Printf("Cleanup: failed to read %q: %v", path, err)
			return nil
		}
		if empty {
			emptyDirs = append(emptyDirs, path)
		}
		return nil
	})

	return emptyDirs, err
}

// removeAll deletes the given paths unless dryRun is set, and returns how many
// were actually removed.
func removeAll(paths []string, kind string, dryRun bool) int {
	removed := 0
	for _, path := range paths {
		if dryRun {
			log.Printf("Dry run: Would remove %s %q", kind, path)
			continue
		}
		if err := os.Remove(path); err != nil {
			log.Printf("Failed to remove %s %q: %v", kind, path, err)
			continue
		}
		log.Printf("Removed %s %q", kind, path)
		removed++
	}
	return removed
}