// Command listend mirrors a phone's podcast subscriptions, works out what to skip in them, and
// serves both to the phone.
//
// It exists because detection cannot work on the phone. The intro/outro search needs several
// episodes' audio compared at once, which needs more memory than an Android process gets and
// more time than WorkManager's ten-minute ceiling allows -- and the attempt burned the battery
// while confirming nothing. Here the same work is unremarkable.
//
// It serves the audio as well as the offsets, which is not a convenience. A show using dynamic
// ad insertion returns a different file on every request, so an offset measured on one copy
// points at the wrong moment in another. The phone playing the very bytes the detector examined
// is what makes the numbers correct.
package main

import (
	"context"
	"errors"
	"flag"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"os/signal"
	"path/filepath"
	"sync"
	"syscall"
	"time"

	"github.com/terst/listen/server/internal/api"
	"github.com/terst/listen/server/internal/decode"
	"github.com/terst/listen/server/internal/feed"
	"github.com/terst/listen/server/internal/mirror"
	"github.com/terst/listen/server/internal/safefetch"
	"github.com/terst/listen/server/internal/signature"
	"github.com/terst/listen/server/internal/skips"
	"github.com/terst/listen/server/internal/store"
	"github.com/terst/listen/server/internal/transcribe"
	"github.com/terst/listen/server/internal/transcript"
	"github.com/terst/listen/server/internal/words"
)

// minRecurringCorpus is how many of a show's episodes must have words before what they share
// means anything. Three: the target and the two others a stretch has to turn up in.
const minRecurringCorpus = 3

func main() {
	var (
		addr      = flag.String("addr", "127.0.0.1:38090", "address to listen on")
		dataDir   = flag.String("data", "/site/listen.terst.org/data", "where state and audio live")
		keep      = flag.Int("keep", 8, "episodes to mirror per show")
		pollEvery = flag.Duration("poll", 2*time.Hour, "how often to refresh every feed")
		tokenFile = flag.String("token-file", "", "file holding the API token (defaults to <data>/token)")
		once      = flag.Bool("once", false, "run one poll cycle and exit, for checking a deployment")

		whisperBin       = flag.String("whisper", "", "whisper-cli binary; empty disables recognition")
		whisperModel     = flag.String("whisper-model", "", "ggml model file for recognition")
		whisperThreads   = flag.Int("whisper-threads", 6, "cores one recognition run may use")
		transcribeBudget = flag.Duration("transcribe-budget", 45*time.Minute,
			"how long each cycle may spend recognising episodes")
		refetchBudget = flag.Duration("refetch-budget", 20*time.Minute,
			"how long each cycle may spend fetching and recognising second copies")
	)
	flag.Parse()

	token, err := readToken(*tokenFile, *dataDir)
	if err != nil {
		log.Fatalf("token: %v", err)
	}

	statePath := filepath.Join(*dataDir, "state.json")
	audioDir := filepath.Join(*dataDir, "audio")
	state, err := store.Load(statePath)
	if err != nil {
		log.Fatalf("state: %v", err)
	}

	signaturePath := filepath.Join(*dataDir, "signatures.json")
	// A damaged library is not worth stopping for: it is evidence that can be gathered again,
	// unlike the state file, which holds the only record of where mirrored audio is.
	signatures, err := signature.Load(signaturePath)
	if err != nil {
		log.Printf("signatures: %v -- starting an empty library", err)
		signatures = signature.NewLibrary()
	}

	decoder := decode.Decoder{}
	transcriptDir := filepath.Join(*dataDir, "transcripts")
	transcripts := transcript.Store{Dir: transcriptDir}
	secondCopyDir := filepath.Join(*dataDir, "work", "second")
	worker := &worker{
		state:         state,
		statePath:     statePath,
		signatures:    signatures,
		signaturePath: signaturePath,
		keep:          *keep,
		mirror:        &mirror.Mirror{AudioDir: audioDir, Probe: decoder},
		decoder:       decoder,
		words: &words.Pipeline{
			State:       state,
			AudioDir:    audioDir,
			Transcripts: transcripts,
			Harvester:   transcript.Harvester{Store: transcripts},
			Recogniser: transcribe.Whisper{
				Binary:  *whisperBin,
				Model:   *whisperModel,
				Threads: *whisperThreads,
				// Beside the audio rather than in /tmp: a decoded hour is 115MB of PCM, and
				// the data directory is the volume with room for it.
				WorkDir: filepath.Join(*dataDir, "work"),
			},
			SecondCopies:     &mirror.Mirror{AudioDir: secondCopyDir, Probe: decoder},
			SecondCopyDir:    secondCopyDir,
			RecogniseBudget:  *transcribeBudget,
			SecondCopyBudget: *refetchBudget,
		},
		requests: make(chan request, 64),
		wakeups:  make(chan struct{}, 1),
	}
	// Wired after the worker exists, because it is the worker's own save. The pipeline runs
	// for tens of minutes at a time and persists as each episode finishes, so a restart in
	// the middle keeps the ones already done.
	worker.words.Save = worker.save

	server := api.New(state, worker, api.Options{
		Token:         token,
		AudioDir:      audioDir,
		TranscriptDir: transcriptDir,
		// Persist the list the moment it arrives, then poll.
		//
		// Both halves were learned the hard way. The list is the one thing the server cannot
		// rediscover, and it used to live only in memory until a whole poll cycle completed --
		// which for 61 feeds is hours -- so a restart in the meantime lost every subscription
		// the phone had sent. And acting on it at once matters because waiting for the next
		// interval leaves a newly-subscribed show with nothing for hours.
		OnSubscriptions: func([]string) {
			worker.save()
			worker.wake()
		},
		// Same reasoning, one step smaller: a label is a tap the listener made once, and it
		// should not sit in memory until the end of a poll cycle. No wake -- nothing acts on a
		// label yet, it is corpus for the signature library.
		OnLabels: worker.save,
	})

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

	if *once {
		worker.words.Reclaim()
		worker.words.Rederive()
		worker.pollAll(ctx)
		// Recognition too, so a deployment check exercises the expensive path rather than
		// discovering it is misconfigured hours later in a log nobody is reading.
		worker.words.Recognise(ctx)
		worker.words.SecondCopy(ctx)
		if err := state.Save(statePath); err != nil {
			log.Fatalf("save: %v", err)
		}
		log.Printf("one cycle complete")
		return
	}

	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		worker.run(ctx, *pollEvery)
	}()

	httpServer := &http.Server{
		Addr:              *addr,
		Handler:           server.Handler(),
		ReadHeaderTimeout: 20 * time.Second,
	}
	go func() {
		<-ctx.Done()
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
		defer cancel()
		_ = httpServer.Shutdown(shutdownCtx)
	}()

	log.Printf("listening on %s, data in %s, keeping %d episodes per show", *addr, *dataDir, *keep)
	if err := httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
		log.Fatalf("http: %v", err)
	}
	wg.Wait()
	if err := state.Save(statePath); err != nil {
		log.Printf("save on exit: %v", err)
	}
}

// request is an episode the phone asked for that the server did not have.
type request struct {
	feedURL, guid, enclosureURL, title string
}

type worker struct {
	state     *store.State
	statePath string
	// signatures is the library the shadow run learns into. Nothing it holds is published yet.
	signatures    *signature.Library
	signaturePath string
	keep          int
	mirror        *mirror.Mirror
	decoder       decode.Decoder
	// transcripts holds each episode's words, and harvester takes the publishers' own where
	// they describe the copy that was mirrored. See transcript.Fits for when they do not.
	// words is the transcript side: taking publishers' transcripts, recognising the rest, and
	// working out what to skip from what was said. It lives in its own package because none
	// of it could be tested here.
	words    *words.Pipeline
	requests chan request
	wakeups  chan struct{}
}

// wake asks for a poll now. Coalescing, so a burst of subscription updates is one poll.
func (w *worker) wake() {
	select {
	case w.wakeups <- struct{}{}:
	default:
	}
}

// Request implements api.Fetcher. Non-blocking: a full queue drops the request rather than
// stalling the phone, and the next poll will pick the episode up anyway.
func (w *worker) Request(feedURL, guid, enclosureURL, title string) {
	select {
	case w.requests <- request{feedURL, guid, enclosureURL, title}:
	default:
		log.Printf("request queue full, dropping %s %s", feedURL, guid)
	}
}

func (w *worker) run(ctx context.Context, every time.Duration) {
	w.words.Reclaim()
	w.words.Rederive()
	w.save()
	w.pollAll(ctx)
	w.save()
	w.words.Recognise(ctx)
	w.words.SecondCopy(ctx)

	ticker := time.NewTicker(every)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			w.pollAll(ctx)
			w.save()
			// Only on the timer. A wakeup is the phone pushing a subscription list and
			// wanting that show polled now; following it with three quarters of an hour of
			// recognition would make a new subscription feel broken.
			w.words.Recognise(ctx)
			w.words.SecondCopy(ctx)
		case <-w.wakeups:
			w.pollAll(ctx)
			w.save()
		case req := <-w.requests:
			w.serveRequest(ctx, req)
			w.save()
		}
	}
}

// drainRequests serves everything the phone has asked for, without waiting.
//
// Bounds how long an on-demand episode waits at one feed rather than one cycle. It does not make
// the wait short -- a feed carrying detection takes minutes -- but the phone falls back to the
// publisher meanwhile, so the cost of waiting is skip offsets rather than playback.
func (w *worker) drainRequests(ctx context.Context) {
	for {
		if ctx.Err() != nil {
			return
		}
		select {
		case req := <-w.requests:
			w.serveRequest(ctx, req)
			w.save()
		default:
			return
		}
	}
}

func (w *worker) save() {
	if err := w.state.Save(w.statePath); err != nil {
		log.Printf("save: %v", err)
	}
}

// pollAll refreshes every subscribed feed, then mirrors and detects what changed.
func (w *worker) pollAll(ctx context.Context) {
	// Topical shows first and archive last, so a five-hour back-catalogue episode cannot take
	// a cycle's recognition budget before the morning news has been looked at.
	feeds := w.state.ByCategory(w.state.Feeds())
	if len(feeds) == 0 {
		log.Printf("no subscriptions yet; waiting for the phone to send its list")
		return
	}
	for _, feedURL := range feeds {
		if ctx.Err() != nil {
			return
		}
		// Before each feed, not only between cycles. The poll loop and the request queue share
		// one goroutine, so an episode the phone asked for used to wait out the entire cycle --
		// hours across 42 shows, during which a tap on an unmirrored episode did nothing at all.
		w.drainRequests(ctx)
		w.pollOne(ctx, feedURL)
		// After every feed rather than at the end of the cycle: a cycle over 61 feeds takes
		// hours, and a restart partway through used to discard all of it.
		w.save()
	}
	// Reclaim the audio of shows no longer subscribed to.
	//
	// Against the list as it stands *now*, not the snapshot this cycle began with. A cycle over
	// 61 feeds takes hours; re-applying a stale snapshot would silently undo a subscription the
	// phone sent during it, and then delete the audio of every show the stale list omitted.
	current := make(map[string]bool)
	for _, url := range w.state.Feeds() {
		current[url] = true
	}
	for _, show := range w.state.Shows() {
		if current[show.FeedURL] {
			continue
		}
		if dropped := w.state.Forget(show.FeedURL); dropped != nil {
			for _, e := range dropped.Episodes {
				_ = w.mirror.Prune(e)
			}
			log.Printf("%s: unsubscribed, reclaimed %d episodes", show.FeedURL, len(dropped.Episodes))
		}
	}
	w.save()
}

func (w *worker) pollOne(ctx context.Context, feedURL string) {
	raw, err := fetchFeed(ctx, feedURL)
	if err != nil {
		w.noteFeedError(feedURL, err)
		return
	}
	channel, err := feed.Parse(raw)
	if err != nil {
		w.noteFeedError(feedURL, err)
		return
	}

	show := w.state.Merge(feedURL, channel.Title, channel.Episodes)
	show.LastPolledUnix = time.Now().Unix()
	show.LastError = ""

	kept, pruned := show.Partition(w.keep, time.Now().Unix())
	for _, e := range pruned {
		if e.Mirrored() {
			_ = w.mirror.Prune(e)
		}
		// The words go with the audio -- see words.Pipeline.Forget.
		w.words.Forget(e)
	}

	fetched := 0
	for _, e := range kept {
		if ctx.Err() != nil {
			return
		}
		if e.Mirrored() {
			continue
		}
		if err := w.mirror.Fetch(ctx, e); err != nil {
			log.Printf("%s: %v", channel.Title, err)
			continue
		}
		fetched++
	}

	if fetched > 0 {
		log.Printf("%s: mirrored %d new episode(s)", channel.Title, fetched)
	}
	// Before anything else looks at this show: an episode just re-mirrored may already have
	// its words on disk from before its record was lost, and recognising it again would cost
	// five minutes for text the server is holding.
	w.words.Reclaim()
	w.words.Harvest(ctx, show, kept)
	w.words.Recurring(show)
	w.detect(ctx, show, kept)
}

// detect runs the intro/outro search over a show's mirrored episodes and records the result.
//
// Every mirrored episode at once, which is the whole reason this is a server: the phone could
// only hold one episode's clip, so it needed a rolling state machine that accumulated agreement
// over weeks and lost it whenever retention reclaimed the file it was holding.
func (w *worker) detect(ctx context.Context, show *store.Show, candidates []*store.Episode) {
	var episodes []skips.Episode
	var ids []string
	byID := map[string]*store.Episode{}
	for _, e := range candidates {
		if !e.Mirrored() {
			continue
		}
		episodes = append(episodes, skips.Episode{
			ID:                 e.ID,
			Path:               filepath.Join(w.mirror.AudioDir, e.Path),
			DeclaredDurationMs: e.DeclaredDurationMs,
		})
		ids = append(ids, e.ID)
		byID[e.ID] = e
	}
	if len(episodes) < skips.DefaultConfirmThreshold || ctx.Err() != nil {
		return
	}
	if !show.NeedsDetection(ids) {
		return
	}

	started := time.Now()
	result, err := skips.Detect(ctx, w.decoder, episodes, skips.DefaultConfig())
	if err != nil {
		log.Printf("%s: detection failed: %v", show.Title, err)
		return
	}

	now := time.Now().Unix()
	for _, o := range result.Offsets {
		if e := byID[o.EpisodeID]; e != nil {
			e.DetectedAtUnix = now
		}
	}

	// Reported, deliberately not published.
	//
	// Intro detection assumes the agreeing episodes were fetched independently. Mirroring eight
	// of them within one minute means they were not, and it showed: the segment confirmed on 8
	// of 8 Radiolab episodes was searched for in an ad-free copy of the same episode fetched
	// hours earlier and scored r = 0.0621. It was an advertisement. Until the agreeing episodes
	// are required to span more than one fetch occasion these numbers cannot be trusted, and
	// the advertising they keep finding is better served by the scan that measures itself
	// against the declared-versus-served length gap.
	log.Printf("%s: intro on %d/%d%s, outro on %d/%d%s (not published -- see detectAds), took %s",
		show.Title,
		result.IntroConfirmedOn, len(episodes), checkNote(result.IntroCheck),
		result.OutroConfirmedOn, len(episodes), checkNote(result.OutroCheck),
		time.Since(started).Round(time.Second))

	if w.detectAds(ctx, show, episodes, byID) {
		show.RecordDetected(ids)
		// After the batch is recorded, so the shadow run's occasion names the batch it
		// learned from. It publishes nothing -- see runShadow.
		w.runShadow(ctx, show, candidates)
	}
}

// detectAds finds the advertising a batch of episodes shares, which is what simultaneous
// mirroring actually makes detectable.
//
// Separate from intro detection, and trusted differently. Intro detection assumes the episodes
// were fetched independently; mirroring eight of them in one minute means they were not, which
// is how it came to confirm an advertisement as an intro. This uses that dependence rather than
// being defeated by it -- and reports what it found against the declared-versus-served length
// gap, a figure it did not produce and which can therefore contradict it.
// Reports whether the scan completed, so an interrupted or failed pass is examined again rather
// than written off as done.
func (w *worker) detectAds(
	ctx context.Context,
	show *store.Show,
	episodes []skips.Episode,
	byID map[string]*store.Episode,
) bool {
	if ctx.Err() != nil {
		return false
	}
	started := time.Now()
	ads, err := skips.DetectAds(ctx, w.decoder, episodes, skips.DefaultAdConfig())
	if err != nil {
		log.Printf("%s: ad scan failed: %v", show.Title, err)
		return false
	}

	var withBreaks, refused int
	var foundTotal, insertedTotal int64
	for _, ep := range episodes {
		e := byID[ep.ID]
		if e == nil {
			continue
		}
		acc := ads.Accounting[ep.ID]
		e.FoundAdMs, e.InsertedAdMs = acc.FoundMs, acc.InsertedMs
		e.AdsRefused = ads.Refused[ep.ID]
		e.Breaks = nil
		for _, br := range ads.Breaks[ep.ID] {
			e.Breaks = append(e.Breaks, store.Break{
				StartMs:  br.StartMs,
				EndMs:    br.EndMs,
				Partners: br.Partners,
				Kind:     string(br.Kind),
			})
		}
		if len(e.Breaks) > 0 {
			withBreaks++
		}
		if e.AdsRefused {
			refused++
		}
		foundTotal += acc.FoundMs
		insertedTotal += acc.InsertedMs
	}

	log.Printf("%s: ads on %d/%d episode(s)%s, accounted %s of %s inserted, took %s",
		show.Title, withBreaks, len(episodes), refusedNote(refused),
		clock(foundTotal), clock(insertedTotal), time.Since(started).Round(time.Second))
	return true
}

func refusedNote(refused int) string {
	if refused == 0 {
		return ""
	}
	return fmt.Sprintf(" (%d refused as too extensive to be advertising)", refused)
}

func clock(ms int64) string {
	total := ms / 1000
	return fmt.Sprintf("%d:%02d", total/60, total%60)
}

// checkNote reports a withheld result loudly. A confirmation that could not verify itself is
// the case where publishing offsets would cut the wrong audio.
func checkNote(r skips.CheckReport) string {
	if r.Pairs == 0 || r.OK {
		return ""
	}
	return fmt.Sprintf(" (SELF-CHECK FAILED %s vs %s at r=%.3f, offsets withheld)", r.WorstA, r.WorstB, r.WorstScore)
}

// serveRequest handles an episode the phone wanted that was not mirrored.
func (w *worker) serveRequest(ctx context.Context, req request) {
	e := w.state.Match(req.feedURL, req.guid, req.enclosureURL, req.title)
	if e == nil {
		// Not in the feed as last polled. Register it from what the phone told us, so an
		// episode older than the mirrored window is still playable with offsets.
		if req.enclosureURL == "" {
			return
		}
		w.state.Merge(req.feedURL, "", []feed.Episode{{
			GUID:          req.guid,
			Title:         req.title,
			EnclosureURL:  req.enclosureURL,
			PublishedUnix: time.Now().Unix(),
		}})
		e = w.state.Match(req.feedURL, req.guid, req.enclosureURL, req.title)
		if e == nil {
			return
		}
	}
	// The api marks episodes it can see; this covers the one it could not, which was registered
	// from what the phone told us a moment ago. Without it the newly-known episode has no
	// record of being wanted and the next poll prunes what was just fetched for it.
	e.Requested(time.Now().Unix())
	if e.Mirrored() {
		return
	}
	if err := w.mirror.Fetch(ctx, e); err != nil {
		log.Printf("on-demand fetch %s: %v", e.ID, err)
		return
	}
	log.Printf("on-demand: mirrored %s", e.Title)

	if show := w.state.Show(req.feedURL); show != nil {
		kept, _ := show.Partition(w.keep, time.Now().Unix())
		w.detect(ctx, show, kept)
	}
}

func (w *worker) noteFeedError(feedURL string, err error) {
	log.Printf("%s: %v", feedURL, err)
	if show := w.state.Show(feedURL); show != nil {
		show.LastError = err.Error()
		show.LastPolledUnix = time.Now().Unix()
	}
}

func fetchFeed(ctx context.Context, url string) ([]byte, error) {
	ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
	defer cancel()
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
	if err != nil {
		return nil, err
	}
	req.Header.Set("User-Agent", "listen-server/1.0 (+https://terst.org)")
	// A feed URL is whatever the phone sent, so the same rule applies as to everything else
	// this server fetches: it may only reach a public address.
	resp, err := safefetch.Client(0).Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("%s returned %s", url, resp.Status)
	}
	// Feeds are large -- a real subscription produced 2753 episodes -- but not unbounded.
	return io.ReadAll(io.LimitReader(resp.Body, 64<<20))
}

// readToken loads the API token, generating one on first run.
func readToken(tokenFile, dataDir string) (string, error) {
	if tokenFile == "" {
		tokenFile = filepath.Join(dataDir, "token")
	}
	if raw, err := os.ReadFile(tokenFile); err == nil {
		token := trimSpace(string(raw))
		if token == "" {
			return "", fmt.Errorf("%s is empty", tokenFile)
		}
		return token, nil
	} else if !os.IsNotExist(err) {
		return "", err
	}

	token, err := generateToken()
	if err != nil {
		return "", err
	}
	if err := os.MkdirAll(filepath.Dir(tokenFile), 0o755); err != nil {
		return "", err
	}
	if err := os.WriteFile(tokenFile, []byte(token+"\n"), 0o600); err != nil {
		return "", err
	}
	log.Printf("generated a new API token in %s", tokenFile)
	return token, nil
}

func trimSpace(s string) string {
	start, end := 0, len(s)
	for start < end && (s[start] == ' ' || s[start] == '\n' || s[start] == '\r' || s[start] == '\t') {
		start++
	}
	for end > start && (s[end-1] == ' ' || s[end-1] == '\n' || s[end-1] == '\r' || s[end-1] == '\t') {
		end--
	}
	return s[start:end]
}