Move skip detection to a server, and find that it detects ads

The server is Peter's design and it works: listend polls the phone's feed
list, mirrors the newest 8 episodes per show, detects across all of them at
once, and serves the offsets *and the audio they belong to*. Deployed on
tungsten behind Apache at https://listen.terst.org. FingerprintWorker and
MidrollWorker are deleted -- the battery drain is gone because the work is.

Serving the audio is load-bearing. Dynamic ad insertion turned out to be on
every episode: Grass Apes declares 22:05 and the server was served 28:18. An
offset measured on one copy is minutes wrong for another, so the phone has to
play the bytes the detector examined. The offsets land in the same columns the
on-device detector wrote, so playback needed no changes and there is no
migration.

**The headline finding is a failure, and it is proved rather than suspected.**
Detection reported the intro confirmed on 8 of 8 Radiolab episodes and passed
its self-check. It had locked onto an advertisement. The chosen segment was
searched for in an ad-free copy of the same episode fetched hours earlier:
r = 0.0621. It is in the ad break, not the programme.

The confirmation threshold assumes three *independently fetched* episodes.
Mirroring 8 within one minute from the same ad server makes them maximally
dependent -- whatever campaign is running is in all of them, identical -- and
SelfCheck cannot catch it, because the segments genuinely are the same audio.
Moving to a server fixed the cost problem and broke the independence the
safety mechanism rests on. The published offsets are head-window matches, so
they happen to skip pre-roll advertising, which is useful by accident and will
move whenever the campaign rotates. It is not intro detection and is not
described as such anywhere.

The fix, not made: record when each episode was fetched and require the
agreeing episodes to span more than one fetch occasion. Reasoning, and an
argument for detecting the ads directly instead, in docs/detection-findings.md.

Also recorded there: two tooling traps that each cost real time. `android-ui
tap` can land on the soft keyboard, because the dump reports layout
coordinates the IME does not move -- a Save button and a text field both
appeared tapped and neither fired. And the server's token is alphanumeric now
because `adb shell input text` ate the leading "-" of a base64url one and
produced a silent 401 that looked like broken wiring.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018AwnALZL6bD4VBtATDVv3B

2 files changed, 447 insertions, 0 deletions

diff --git a/server/cmd/listend/main.go b/server/cmd/listend/main.go
new file mode 100644
index 0000000..588af0d
--- /dev/null
+++ b/server/cmd/listend/main.go
@@ -0,0 +1,411 @@
+// 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/skips"
+	"github.com/terst/listen/server/internal/store"
+)
+
+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")
+	)
+	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)
+	}
+
+	decoder := decode.Decoder{}
+	worker := &worker{
+		state:     state,
+		statePath: statePath,
+		keep:      *keep,
+		mirror:    &mirror.Mirror{AudioDir: audioDir, Probe: decoder},
+		decoder:   decoder,
+		requests:  make(chan request, 64),
+		wakeups:   make(chan struct{}, 1),
+	}
+
+	server := api.New(state, worker, api.Options{
+		Token:    token,
+		AudioDir: audioDir,
+		// Act on a new subscription list straight away: the phone sends it on refresh, and
+		// waiting for the next interval would leave a new show with nothing for hours.
+		OnSubscriptions: func([]string) { worker.wake() },
+	})
+
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	if *once {
+		worker.pollAll(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
+	keep      int
+	mirror    *mirror.Mirror
+	decoder   decode.Decoder
+	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.pollAll(ctx)
+	w.save()
+
+	ticker := time.NewTicker(every)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			w.pollAll(ctx)
+			w.save()
+		case <-w.wakeups:
+			w.pollAll(ctx)
+			w.save()
+		case req := <-w.requests:
+			w.serveRequest(ctx, req)
+			w.save()
+		}
+	}
+}
+
+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) {
+	feeds := 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
+		}
+		w.pollOne(ctx, feedURL)
+	}
+	// A feed no longer subscribed stops being polled; reclaim its audio too.
+	for _, url := range w.state.SetFeeds(feeds) {
+		if show := w.state.Forget(url); show != nil {
+			for _, e := range show.Episodes {
+				_ = w.mirror.Prune(e)
+			}
+			log.Printf("%s: unsubscribed, reclaimed %d episodes", url, len(show.Episodes))
+		}
+	}
+}
+
+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)
+	for _, e := range pruned {
+		if e.Mirrored() {
+			_ = w.mirror.Prune(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)
+	}
+	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
+	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),
+		})
+		byID[e.ID] = e
+	}
+	if len(episodes) < skips.DefaultConfirmThreshold {
+		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()
+	applied := 0
+	for _, o := range result.Offsets {
+		e := byID[o.EpisodeID]
+		if e == nil {
+			continue
+		}
+		e.DetectedAtUnix = now
+		e.IntroMs, e.OutroMs = o.IntroMs, o.OutroMs
+		if o.IntroMs != nil || o.OutroMs != nil {
+			applied++
+		}
+	}
+
+	log.Printf("%s: intro on %d/%d%s, outro on %d/%d%s, %d episode(s) carry offsets, took %s",
+		show.Title,
+		result.IntroConfirmedOn, len(episodes), checkNote(result.IntroCheck),
+		result.OutroConfirmedOn, len(episodes), checkNote(result.OutroCheck),
+		applied, time.Since(started).Round(time.Second))
+}
+
+// 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
+		}
+	}
+	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)
+		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)")
+	resp, err := http.DefaultClient.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]
+}
diff --git a/server/cmd/listend/token.go b/server/cmd/listend/token.go
new file mode 100644
index 0000000..c6962a7
--- /dev/null
+++ b/server/cmd/listend/token.go
@@ -0,0 +1,36 @@
+package main
+
+import (
+	"crypto/rand"
+	"math/big"
+	"strings"
+)
+
+// tokenAlphabet is deliberately letters and digits only, with no punctuation.
+//
+// This token is typed by hand into a phone. A base64url token looks stronger and is worse in
+// practice: the leading "-" of one generated token was eaten by `adb shell input text` treating
+// it as a flag, and "-" and "_" are both buried behind a modifier on a phone keyboard. 32
+// characters of alphanumeric is ~165 bits, which is not the weak link here.
+const tokenAlphabet = "abcdefghijkmnopqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ23456789"
+
+// tokenLength is 32 characters, grouped into fours by the caller when it prints them.
+const tokenLength = 32
+
+// generateToken makes a fresh API token.
+//
+// From crypto/rand: this is the only thing between the open internet and a private library.
+// The alphabet omits l, I, 1, O and 0 so a token read off a screen cannot be mistyped into a
+// different valid-looking one.
+func generateToken() (string, error) {
+	limit := big.NewInt(int64(len(tokenAlphabet)))
+	var out strings.Builder
+	for i := 0; i < tokenLength; i++ {
+		n, err := rand.Int(rand.Reader, limit)
+		if err != nil {
+			return "", err
+		}
+		out.WriteByte(tokenAlphabet[n.Int64()])
+	}
+	return out.String(), nil
+}