record-asr/internal/server/server.go

// Package server implements the HTTP API: password auth, the task endpoints
// (upload / list / get / delete / retry / skip), and the background worker
// that drains the queue with automatic retry-on-error.
package server

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"sync"
	"time"

	"git.misaka.ren/m1saka/record-asr/internal/config"
	"git.misaka.ren/m1saka/record-asr/internal/models"
	"git.misaka.ren/m1saka/record-asr/internal/store"
	"git.misaka.ren/m1saka/record-asr/internal/transcribe"
)

// Server wires the store, transcribe pipeline, and model manager behind an
// authenticated HTTP API.
type Server struct {
	cfg     config.Config
	store   *store.Store
	pipe    *transcribe.Pipeline
	manager *models.Manager
	// concSem caps simultaneously in-flight synchronous transcriptions on
	// the OpenAI endpoint, preventing onnxruntime thread contention / OOM
	// under bursts. The background workers are NOT gated by this semaphore;
	// they are bounded separately by ASR_WORKERS, so the task queue is
	// never starved by a flood of OpenAI requests.
	concSem chan struct{}
}

// New assembles a Server from its dependencies.
func New(cfg config.Config, st *store.Store, pipe *transcribe.Pipeline, mgr *models.Manager) *Server {
	return &Server{
		cfg:     cfg,
		store:   st,
		pipe:    pipe,
		manager: mgr,
		concSem: make(chan struct{}, cfg.MaxConcurrent),
	}
}

// Router builds the HTTP mux. All routes are registered with explicit
// method + path patterns on one mux (Go 1.22+). Auth is then applied as an
// outer middleware that only enforces the password for /api/* paths, leaving
// /healthz and / public so monitoring and the landing page work token-free.
func (s *Server) Router() http.Handler {
	mux := http.NewServeMux()

	// Public endpoints: the web console (UI calls the authed /api/* routes
	// itself) and the health check.
	mux.HandleFunc("GET /", s.handleUI)
	mux.HandleFunc("GET /healthz", s.handleHealthz)

	// Task API (guarded by the outer authMiddleware below).
	mux.HandleFunc("POST /api/tasks", s.handleCreateTask)
	mux.HandleFunc("GET /api/tasks", s.handleListTasks)
	mux.HandleFunc("GET /api/tasks/{id}", s.handleGetTask)
	mux.HandleFunc("DELETE /api/tasks/{id}", s.handleDeleteTask)
	mux.HandleFunc("POST /api/tasks/{id}/retry", s.handleRetryTask)
	mux.HandleFunc("POST /api/tasks/{id}/skip", s.handleSkipTask)
	mux.HandleFunc("GET /api/tasks/{id}/transcript", s.handleTranscript)

	// Resumable chunked uploads (also authed). Lets a large file slip past
	// size-capping proxies (e.g. Cloudflare's 100MB body limit) as many
	// small slices, finalized into a task indistinguishable from /api/tasks.
	mux.HandleFunc("POST /api/uploads", s.handleCreateUpload)
	mux.HandleFunc("POST /api/uploads/{id}", s.handleUploadChunk)
	mux.HandleFunc("POST /api/uploads/{id}/complete", s.handleCompleteUpload)
	mux.HandleFunc("DELETE /api/uploads/{id}", s.handleAbortUpload)

	// OpenAI-compatible synchronous transcription (also authed).
	mux.HandleFunc("POST /v1/audio/transcriptions", s.handleOpenAITranscription)

	return s.authMiddleware(mux)
}

// Run starts the background workers and serves HTTP until ctx is cancelled.
func (s *Server) Run(ctx context.Context) error {
	if s.cfg.AuthEnabled() {
		log.Printf("auth: enabled (password set)")
	} else {
		log.Printf("WARNING: ASR_PASSWORD not set — API is open; set ASR_PASSWORD to require a password")
	}
	log.Printf("concurrency: max %d in-flight sync transcriptions (OpenAI endpoint)", s.cfg.MaxConcurrent)

	var wg sync.WaitGroup
	for i := 0; i < s.cfg.Workers; i++ {
		wg.Add(1)
		go func(n int) {
			defer wg.Done()
			s.runWorker(ctx, n)
		}(i)
	}

	srv := &http.Server{
		Addr:              s.cfg.Addr,
		Handler:           s.Router(),
		ReadHeaderTimeout: 15 * time.Second,
		// Reading a large upload can take a while; bound it generously.
		ReadTimeout:       10 * time.Minute,
		WriteTimeout:      15 * time.Minute,
		IdleTimeout:       120 * time.Second,
	}

	errCh := make(chan error, 1)
	go func() {
		log.Printf("listening on %s", s.cfg.Addr)
		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			errCh <- err
			return
		}
		errCh <- nil
	}()

	select {
	case <-ctx.Done():
	case err := <-errCh:
		if err != nil {
			return err
		}
	}

	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if err := srv.Shutdown(shutdownCtx); err != nil {
		log.Printf("http shutdown: %v", err)
	}
	wg.Wait()
	return nil
}

// runWorker is the queue drain loop. It claims one runnable task at a time,
// transcribes it, and on error requeues it (with backoff, up to MaxRetries)
// or marks it failed — then immediately looks for the next task. This is the
// automatic "retry-or-continue" behavior.
func (s *Server) runWorker(ctx context.Context, n int) {
	log.Printf("worker %d started", n)
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			log.Printf("worker %d stopping", n)
			return
		default:
		}
		t := s.store.Claim(time.Now())
		if t == nil {
			// Nothing ready; wait briefly so we don't spin.
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
			}
			continue
		}
		s.processTask(ctx, t)
	}
}

// processTask runs the pipeline for one claimed task and records the outcome.
func (s *Server) processTask(ctx context.Context, t *store.Task) {
	start := time.Now()
	audioPath := s.store.AudioPath(t)
	log.Printf("[%s] start (attempt %d, %s)", t.ID, t.Attempts, t.OriginalName)

	res, err := s.pipe.Transcribe(audioPath)
	if err != nil {
		log.Printf("[%s] failed: %v", t.ID, err)
		if ferr := s.store.Fail(t.ID, err.Error()); ferr != nil {
			log.Printf("[%s] persist failure: %v", t.ID, ferr)
		}
		// Whether retried or marked failed, the worker continues to the next.
		return
	}
	if perr := s.store.Complete(t.ID, res); perr != nil {
		log.Printf("[%s] persist complete: %v", t.ID, perr)
		return
	}
	log.Printf("[%s] done in %.1fs (%d segments, lang=%s)", t.ID,
		time.Since(start).Seconds(), len(res.Segments), res.Lang)
}

// manager access for handlers (read-only status).
func (s *Server) modelVersion() string {
	if s.manager == nil {
		return "n/a"
	}
	return s.manager.VersionInfo()
}

func (s *Server) dataDir() string { return s.cfg.DataDir }

// renderError writes a small JSON error object with the given status.
func renderError(w http.ResponseWriter, status int, msg string) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(status)
	fmt.Fprintf(w, `{"error":%q}`, msg)
}