path: root/server/internal/skips/ads.go

package skips

import (
	"context"
	"fmt"
	"sort"

	"github.com/terst/listen/server/internal/decode"
	"github.com/terst/listen/server/internal/detect"
)

/*
Detecting the advertising rather than the intro.

The inversion the data forced. Intro detection reported a segment confirmed on 8 of 8 Radiolab
episodes, passed its own self-check, and was wrong: searched for in an ad-free copy of the same
episode fetched hours earlier it scored r = 0.0621. It was in the ad break.

Mirroring 8 episodes within one minute makes them maximally dependent -- the ad server inserts
whatever campaign is running into all of them, identically -- and that destroys the independence
the confirmation threshold needs. No amount of self-checking recovers it, because the segments
genuinely are the same audio.

Turned around, the same property is a reliable advertisement detector: what simultaneous fetches
share *is* the inserted advertising. And unlike intro detection this can be checked against
something the detector did not produce -- the feed declares the ad-free length, the server was
served something longer, and the difference is how much advertising there should be.
*/

// AudioSource is the decoding ad detection needs.
type AudioSource interface {
	// RangeAt decodes mono audio at a chosen sample rate. Zero durationMs means to the end.
	RangeAt(ctx context.Context, path string, startMs, durationMs int64, sampleRate int) (*decode.Clip, error)
	// DurationMs is the file's real length.
	DurationMs(ctx context.Context, path string) (int64, error)
}

// AdBreak is a stretch of one episode that other episodes also contain.
type AdBreak struct {
	StartMs int64
	EndMs   int64
	// Partners is how many other episodes were found to carry this audio.
	Partners int
	// Score is the weakest window agreement across the match, so it describes the whole break.
	Score float64
	// Kind says whether this is inserted advertising or the show's own audio. Set by Classify.
	Kind BreakKind
}

// DurationMs is the break's length.
func (b AdBreak) DurationMs() int64 { return b.EndMs - b.StartMs }

// Accounting is what the scan found against what the length gap says should be there.
//
// The gap is the point: it comes from the feed and the file, not from the detector, so it is the
// one number here that can contradict the scan. Far short means breaks are being missed; far
// over means programme is being matched.
type Accounting struct {
	FoundMs int64
	// InsertedMs is served minus declared, or zero when the feed did not say.
	InsertedMs int64
}

// AdResult is a batch's outcome.
type AdResult struct {
	Breaks     map[string][]AdBreak
	Accounting map[string]Accounting
	// Refused records episodes whose matches were too extensive to be advertising, and which
	// therefore publish nothing.
	Refused map[string]bool
}

// AdConfig governs the scan.
type AdConfig struct {
	// DiscoveryRate is the sample rate whole episodes are compared at. 250Hz turns an hour of
	// audio from 29 million samples into 900 thousand, which is what makes a single
	// cross-correlation over a whole pair of episodes possible.
	DiscoveryRate int
	// LagsPerPair is how many candidate alignments are examined per pair of episodes. An
	// episode carries a handful of ad breaks, and each one shows up as its own alignment.
	LagsPerPair int
	// MaxSkipFraction refuses an episode whose matches cover more than this much of it.
	//
	// The guard that matters most: two episodes of a show can legitimately share a lot -- a
	// re-aired episode, a two-parter with a long recap -- and skipping all of it would eat the
	// programme rather than the advertising.
	MaxSkipFraction float64
	// MaxBreakMs caps a single *match*. Longer than any plausible ad pod means the episodes
	// share programme -- a re-aired episode, a two-parter's recap -- rather than advertising.
	//
	// Deliberately applied before bridging, not after. One enormous match is evidence of shared
	// programme; a chain of ordinary matches a beat apart is evidence of an ad pod, and fusing
	// them must not manufacture a refusal. Radiolab's Grass Apes carries a 3:55 pre-roll and an
	// 0:18 tag two seconds later, and capping the bridged 4:15 would publish nothing for the
	// best-validated episode in the corpus.
	MaxBreakMs int64
	// MergeGapMs bridges matches separated by less than this.
	//
	// The model is the ad pod: a handful of advertisements back to back, with short gaps where
	// agreement lapses between them, and long stretches of real programme between pods. 10s
	// caught the lapses but not the pod -- fragments of one pod separated by more than ten
	// seconds stayed as separate breaks instead of the single break a listener actually
	// experiences. 60s bridges a whole pod's internal gaps while still leaving genuine programme
	// between pods intact; combined with the twelve-second run floor this halved the breaks
	// found corpus-wide (1091 to 539) while *improving* agreement with the declared-versus-served
	// length gap (31% to 34% of episodes within 25%), which is the signal that what was removed
	// was noise rather than real advertising.
	MergeGapMs int64
	// Runs is what counts as agreement within an alignment.
	Runs detect.RunOptions
}

// DefaultAdConfig is the production configuration.
func DefaultAdConfig() AdConfig {
	return AdConfig{
		DiscoveryRate:   250,
		LagsPerPair:     8,
		MaxSkipFraction: 0.35,
		MaxBreakMs:      240_000,
		MergeGapMs:      60_000,
		Runs:            detect.DefaultRunOptions(),
	}
}

// DetectAds finds what a batch of episodes share, which for simultaneously-mirrored episodes is
// the advertising inserted into them.
//
// Pairwise, because an ad need not be in every episode -- campaigns rotate within a batch, and a
// break present in two episodes is still a break. Every pair costs one cross-correlation over
// the whole decimated episode, not a correlation per window position: the sliding search that
// replaces measured 199 seconds for a 240-second clip and could never have run over whole
// episodes.
func DetectAds(ctx context.Context, src AudioSource, episodes []Episode, cfg AdConfig) (*AdResult, error) {
	result := &AdResult{
		Breaks:     map[string][]AdBreak{},
		Accounting: map[string]Accounting{},
		Refused:    map[string]bool{},
	}
	if len(episodes) < 2 {
		return result, nil
	}

	rate := cfg.DiscoveryRate
	if rate <= 0 {
		rate = DefaultAdConfig().DiscoveryRate
	}

	type loaded struct {
		episode  Episode
		samples  []float64
		servedMs int64
	}
	clips := make([]loaded, 0, len(episodes))
	for _, ep := range episodes {
		servedMs, err := src.DurationMs(ctx, ep.Path)
		if err != nil {
			return nil, fmt.Errorf("episode %s: %w", ep.ID, err)
		}
		clip, err := src.RangeAt(ctx, ep.Path, 0, 0, rate)
		if err != nil {
			return nil, fmt.Errorf("episode %s: %w", ep.ID, err)
		}
		clips = append(clips, loaded{episode: ep, samples: clip.Samples, servedMs: servedMs})
	}

	// found[episode index] accumulates every match from every partner, before merging.
	found := make([][]AdBreak, len(clips))
	for i := 0; i < len(clips); i++ {
		for j := i + 1; j < len(clips); j++ {
			a, b := clips[i].samples, clips[j].samples
			for _, lag := range detect.BestLags(a, b, cfg.LagsPerPair) {
				for _, run := range detect.MatchedRuns(a, b, lag, cfg.Runs) {
					found[i] = append(found[i], breakOf(run.StartA, run.EndA, run.Score, rate))
					found[j] = append(found[j], breakOf(run.StartB, run.StartB+run.Length(), run.Score, rate))
				}
			}
		}
	}

	for i, clip := range clips {
		// Merged without judgement first, so the accounting describes what was actually found
		// and a refusal is recorded rather than silently leaving an episode with nothing. An
		// episode that publishes no breaks because its matches were implausible is a different
		// state from one where nothing matched, and the two looking identical is the mistake
		// the app's own skip readout already made once.
		// A single match longer than any real ad pod is shared programme -- a recap, or a
		// repeat -- and cutting it would remove the episode rather than the advertising. Judged
		// on the matches, before bridging fuses neighbours into something longer.
		kept := make([]AdBreak, 0, len(found[i]))
		for _, br := range found[i] {
			if cfg.MaxBreakMs > 0 && br.DurationMs() > cfg.MaxBreakMs {
				result.Refused[clip.episode.ID] = true
				continue
			}
			kept = append(kept, br)
		}

		matched := merge(kept, cfg.MergeGapMs)
		total := int64(0)
		for _, br := range matched {
			total += br.DurationMs()
		}

		inserted := int64(0)
		if clip.episode.DeclaredDurationMs > 0 && clip.servedMs > clip.episode.DeclaredDurationMs {
			inserted = clip.servedMs - clip.episode.DeclaredDurationMs
		}
		result.Accounting[clip.episode.ID] = Accounting{FoundMs: total, InsertedMs: inserted}

		// Too much of the episode matched to be advertising at all.
		if clip.servedMs > 0 && float64(total) > cfg.MaxSkipFraction*float64(clip.servedMs) {
			result.Refused[clip.episode.ID] = true
			continue
		}

		if len(matched) > 0 {
			result.Breaks[clip.episode.ID] = Classify(matched, clip.servedMs, clip.episode.DeclaredDurationMs)
		}
	}
	return result, nil
}

func breakOf(startSamples, endSamples int, score float64, rate int) AdBreak {
	return AdBreak{
		StartMs:  int64(startSamples) * 1000 / int64(rate),
		EndMs:    int64(endSamples) * 1000 / int64(rate),
		Partners: 1,
		Score:    score,
	}
}

// merge folds matches into single breaks, chronologically, bridging gaps up to gapMs.
//
// Two things at once. The same advertisement is found once per partner that carries it, so
// without merging an ad in three episodes becomes three identical breaks. And a pod of several
// advertisements is found as several matches with brief lapses between them, which bridging
// turns back into the one stretch a listener experiences.
//
// The partner count is what survives: it says how much agreement a break rests on.
func merge(raw []AdBreak, gapMs int64) []AdBreak {
	if len(raw) == 0 {
		return nil
	}
	sort.Slice(raw, func(i, j int) bool { return raw[i].StartMs < raw[j].StartMs })

	out := make([]AdBreak, 0, len(raw))
	current := raw[0]
	for _, next := range raw[1:] {
		if next.StartMs <= current.EndMs+gapMs {
			if next.EndMs > current.EndMs {
				current.EndMs = next.EndMs
			}
			current.Partners++
			if next.Score < current.Score {
				current.Score = next.Score
			}
			continue
		}
		out = append(out, current)
		current = next
	}
	return append(out, current)
}