Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
27 kB · 669 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670package viewlog
import ( "context" "sort" "strings" "testing" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "github.com/stretchr/testify/require" "stream.place/streamplace/pkg/comatproto"
"stream.place/streamplace/pkg/blob" "stream.place/streamplace/pkg/vod")
// fixtureTrackRefs returns the strongRefs that pair with the fixture// metafile, keyed by in-container trackId. Same shape the real// resolver in cmd/streamplace.go produces from MediaTrack rows.func fixtureTrackRefs() map[string]comatproto.RepoStrongRef { return map[string]comatproto.RepoStrongRef{ "1": { LexiconTypeID: "com.atproto.repo.strongRef", Uri: "at://did:plc:alice/place.stream.media.track/track-video-1", Cid: "bafytrack1", }, "2": { LexiconTypeID: "com.atproto.repo.strongRef", Uri: "at://did:plc:alice/place.stream.media.track/track-audio-2", Cid: "bafytrack2", }, }}
// loadStore returns an in-memory-ish FileStore + a writer pointed at// it, sharing the same NodeDID across test cases so the file naming// stays predictable.func newAggTestWriter(t *testing.T, root string, nodeDID string, now time.Time) *Writer { t.Helper() store, err := blob.NewFileStore(root) require.NoError(t, err) w, err := NewWriter(Config{ Store: store, NodeDID: nodeDID, FlushAfter: 1 * time.Hour, Salts: NewSaltManager(newMemSaltStorage()), Now: func() time.Time { return now }, }) require.NoError(t, err) return w}
// runOne drives the writer's flush loop once, then closes.func runOneAndClose(t *testing.T, ctx context.Context, w *Writer) { t.Helper() go w.Run(ctx) require.NoError(t, w.Close())}
func TestAggregateWindowBasicView(t *testing.T) { root := t.TempDir() now := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC)
w := newAggTestWriter(t, root, "did:web:node1", now) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go w.Run(ctx)
const ( videoA = "at://did:plc:alice/place.stream.video/v1" sid1 = "tidsid1" ) // One sid, one video, three segments — should count as one view. w.Log(ctx, Event{Ts: now, Type: EventTypeManifestRequest, VideoURI: videoA, SID: sid1, ManifestKind: ManifestKindMaster}) w.Log(ctx, Event{Ts: now.Add(time.Second), Type: EventTypeSegmentRequest, CID: "bafy", SID: sid1}) w.Log(ctx, Event{Ts: now.Add(2 * time.Second), Type: EventTypeSegmentRequest, CID: "bafy", SID: sid1}) w.Log(ctx, Event{Ts: now.Add(3 * time.Second), Type: EventTypeSegmentRequest, CID: "bafy", SID: sid1}) require.NoError(t, w.Close())
store, err := blob.NewFileStore(root) require.NoError(t, err) res, err := AggregateWindow(context.Background(), store, AggregateInput{ WindowStart: now.Add(-time.Minute), WindowEnd: now.Add(time.Minute), }) require.NoError(t, err) require.Len(t, res.VideoCounts, 1) require.Equal(t, videoA, res.VideoCounts[0].VideoURI) require.Equal(t, int64(1), res.VideoCounts[0].Count)}
func TestAggregateWindowDistinctSids(t *testing.T) { root := t.TempDir() now := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) w := newAggTestWriter(t, root, "did:web:node1", now) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go w.Run(ctx)
const ( videoA = "at://did:plc:alice/place.stream.video/v1" videoB = "at://did:plc:bob/place.stream.video/v2" ) // 3 distinct sids watching videoA, 1 watching videoB. for i, sid := range []string{"sidA1", "sidA2", "sidA3"} { ts := now.Add(time.Duration(i) * time.Second) w.Log(ctx, Event{Ts: ts, Type: EventTypeManifestRequest, VideoURI: videoA, SID: sid}) w.Log(ctx, Event{Ts: ts.Add(time.Millisecond), Type: EventTypeSegmentRequest, CID: "bafyA", SID: sid}) } w.Log(ctx, Event{Ts: now.Add(10 * time.Second), Type: EventTypeManifestRequest, VideoURI: videoB, SID: "sidB1"}) w.Log(ctx, Event{Ts: now.Add(11 * time.Second), Type: EventTypeSegmentRequest, CID: "bafyB", SID: "sidB1"}) require.NoError(t, w.Close())
store, err := blob.NewFileStore(root) require.NoError(t, err) res, err := AggregateWindow(context.Background(), store, AggregateInput{ WindowStart: now.Add(-time.Minute), WindowEnd: now.Add(time.Minute), }) require.NoError(t, err)
got := map[string]int64{} for _, vc := range res.VideoCounts { got[vc.VideoURI] = vc.Count } require.Equal(t, map[string]int64{videoA: 3, videoB: 1}, got)}
func TestAggregateWindowThreshold(t *testing.T) { root := t.TempDir() now := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) w := newAggTestWriter(t, root, "did:web:node1", now) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go w.Run(ctx)
const videoA = "at://did:plc:alice/place.stream.video/v1" // sidA fetched 1 segment, sidB fetched 3 — only sidB clears a // threshold of 2. w.Log(ctx, Event{Ts: now, Type: EventTypeManifestRequest, VideoURI: videoA, SID: "sidA"}) w.Log(ctx, Event{Ts: now.Add(time.Second), Type: EventTypeSegmentRequest, CID: "bafy", SID: "sidA"})
w.Log(ctx, Event{Ts: now.Add(2 * time.Second), Type: EventTypeManifestRequest, VideoURI: videoA, SID: "sidB"}) w.Log(ctx, Event{Ts: now.Add(3 * time.Second), Type: EventTypeSegmentRequest, CID: "bafy", SID: "sidB"}) w.Log(ctx, Event{Ts: now.Add(4 * time.Second), Type: EventTypeSegmentRequest, CID: "bafy", SID: "sidB"}) w.Log(ctx, Event{Ts: now.Add(5 * time.Second), Type: EventTypeSegmentRequest, CID: "bafy", SID: "sidB"}) require.NoError(t, w.Close())
store, err := blob.NewFileStore(root) require.NoError(t, err) res, err := AggregateWindow(context.Background(), store, AggregateInput{ WindowStart: now.Add(-time.Minute), WindowEnd: now.Add(time.Minute), ThresholdSegments: 2, }) require.NoError(t, err) require.Equal(t, int64(1), res.VideoCounts[0].Count, "only sids with ≥ threshold segments count")}
func TestAggregateWindowSegmentsWithoutManifestDropped(t *testing.T) { root := t.TempDir() now := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) w := newAggTestWriter(t, root, "did:web:node1", now) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go w.Run(ctx)
// Segment events for a sid that never made a manifest request — // can't be attributed to a video, so they vanish. w.Log(ctx, Event{Ts: now, Type: EventTypeSegmentRequest, CID: "bafy", SID: "orphan"}) w.Log(ctx, Event{Ts: now.Add(time.Second), Type: EventTypeSegmentRequest, CID: "bafy", SID: "orphan"}) require.NoError(t, w.Close())
store, err := blob.NewFileStore(root) require.NoError(t, err) res, err := AggregateWindow(context.Background(), store, AggregateInput{ WindowStart: now.Add(-time.Minute), WindowEnd: now.Add(time.Minute), }) require.NoError(t, err) require.Empty(t, res.VideoCounts)}
func TestAggregateWindowOutOfWindowEventsExcluded(t *testing.T) { root := t.TempDir() windowStart := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) windowEnd := windowStart.Add(5 * time.Minute)
w := newAggTestWriter(t, root, "did:web:node1", windowStart) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go w.Run(ctx)
const videoA = "at://did:plc:alice/place.stream.video/v1" // Segments before the window are ignored, segments in the window // count, segments after are ignored. The manifest event timestamp // is irrelevant — sid→video carries across the boundary. w.Log(ctx, Event{Ts: windowStart.Add(-time.Hour), Type: EventTypeManifestRequest, VideoURI: videoA, SID: "sidEarly"}) w.Log(ctx, Event{Ts: windowStart.Add(-30 * time.Second), Type: EventTypeSegmentRequest, CID: "bafy", SID: "sidEarly"}) // dropped w.Log(ctx, Event{Ts: windowStart.Add(time.Second), Type: EventTypeSegmentRequest, CID: "bafy", SID: "sidEarly"}) // kept w.Log(ctx, Event{Ts: windowEnd.Add(time.Second), Type: EventTypeSegmentRequest, CID: "bafy", SID: "sidEarly"}) // dropped require.NoError(t, w.Close())
store, err := blob.NewFileStore(root) require.NoError(t, err) res, err := AggregateWindow(context.Background(), store, AggregateInput{ WindowStart: windowStart, WindowEnd: windowEnd, }) require.NoError(t, err) require.Len(t, res.VideoCounts, 1, "the in-window segment still counts (sid→video carried from earlier manifest)") require.Equal(t, int64(1), res.VideoCounts[0].Count)}
func TestAggregateWindowCrossNodeTimeOrdering(t *testing.T) { root := t.TempDir() now := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) // Two nodes, alphabetically reversed from time order: nodeZ writes // the earlier manifest, nodeA writes the later segment. Pure-key // lex sort would read the segment first and miss the attribution. wZ := newAggTestWriter(t, root, "did:web:nodeZ", now) wA := newAggTestWriter(t, root, "did:web:nodeA", now.Add(time.Minute)) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go wZ.Run(ctx) go wA.Run(ctx)
const ( videoA = "at://did:plc:alice/place.stream.video/v1" sid = "wandering-sid" ) wZ.Log(ctx, Event{Ts: now, Type: EventTypeManifestRequest, VideoURI: videoA, SID: sid}) wA.Log(ctx, Event{Ts: now.Add(2 * time.Minute), Type: EventTypeSegmentRequest, CID: "bafy", SID: sid}) require.NoError(t, wZ.Close()) require.NoError(t, wA.Close())
store, err := blob.NewFileStore(root) require.NoError(t, err) res, err := AggregateWindow(context.Background(), store, AggregateInput{ WindowStart: now.Add(-time.Minute), WindowEnd: now.Add(10 * time.Minute), }) require.NoError(t, err) require.Len(t, res.VideoCounts, 1) require.Equal(t, int64(1), res.VideoCounts[0].Count)}
func TestAggregateWindowDeterministicOrder(t *testing.T) { root := t.TempDir() now := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) w := newAggTestWriter(t, root, "did:web:node1", now) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go w.Run(ctx)
for i, video := range []string{ "at://did:plc:c/place.stream.video/v1", "at://did:plc:a/place.stream.video/v1", "at://did:plc:b/place.stream.video/v1", } { sid := video // unique sid per video w.Log(ctx, Event{Ts: now.Add(time.Duration(i) * time.Second), Type: EventTypeManifestRequest, VideoURI: video, SID: sid}) w.Log(ctx, Event{Ts: now.Add(time.Duration(i)*time.Second + time.Millisecond), Type: EventTypeSegmentRequest, CID: "bafy", SID: sid}) } require.NoError(t, w.Close())
store, err := blob.NewFileStore(root) require.NoError(t, err) res, err := AggregateWindow(context.Background(), store, AggregateInput{ WindowStart: now.Add(-time.Minute), WindowEnd: now.Add(time.Minute), }) require.NoError(t, err) got := make([]string, 0, len(res.VideoCounts)) for _, vc := range res.VideoCounts { got = append(got, vc.VideoURI) } // Output is sorted by VideoURI so callers get a stable iteration. require.True(t, sort.StringsAreSorted(got), "VideoCounts should be sorted by URI for stable record rkeys, got %v", got)}
func TestAggregateTaskKeyIsDeterministic(t *testing.T) { start := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) end := start.Add(5 * time.Minute) k1 := AggregateTaskKey(start, end) k2 := AggregateTaskKey(start, end) require.Equal(t, k1, k2, "same window must produce same key (dedups across nodes)") require.NotEqual(t, k1, AggregateTaskKey(start, end.Add(time.Minute)))}
func TestViewCountRkeyShape(t *testing.T) { videoURI := "at://did:plc:alice/place.stream.video/3jw5xvr5gck2a" windowStart := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC)
r, err := viewCountRkey(videoURI, windowStart) require.NoError(t, err) require.Contains(t, r, "-", "rkey is <windowTID>-<videoTID>")
parts := strings.SplitN(r, "-", 2) require.Len(t, parts, 2) // Both halves should be parseable as TIDs (the joined whole is // not — atproto's TID is 13 chars exactly, the joined form has // 27). _, err = syntax.ParseTID(parts[0]) require.NoError(t, err, "window half should be a valid TID") _, err = syntax.ParseTID(parts[1]) require.NoError(t, err, "video half should be a valid TID") require.Equal(t, "3jw5xvr5gck2a", parts[1], "video half is the AT-URI's record key")
// Same inputs → same rkey (idempotent overwrites on rerun). r2, err := viewCountRkey(videoURI, windowStart) require.NoError(t, err) require.Equal(t, r, r2)
// Different window → different rkey. rLater, err := viewCountRkey(videoURI, windowStart.Add(5*time.Minute)) require.NoError(t, err) require.NotEqual(t, r, rLater)}
// fixtureMetafile returns a small two-track metafile suitable for// driving the overlap math. Track 1 is video, track 2 is audio; both// at timescale 1000 so 1 tick == 1 ms (makes test arithmetic trivial).// Three segments per track, contiguous, video bytes laid out before// audio bytes within each segment — same order the writer uses.func fixtureMetafile() *vod.Metafile { return &vod.Metafile{ BlobCID: "bafyfixture", BlobSize: 1500, Tracks: map[string]vod.MetafileTrack{ "1": { Type: "video", Timescale: 1000, Segments: []vod.MetafileSegment{ {Offset: 0, Size: 400, DurationTicks: 2000}, // [0..399] = 2s {Offset: 500, Size: 400, DurationTicks: 2000}, // [500..899] = 2s {Offset: 1000, Size: 400, DurationTicks: 2000}, }, }, "2": { Type: "audio", Timescale: 1000, Segments: []vod.MetafileSegment{ {Offset: 400, Size: 100, DurationTicks: 2000}, // [400..499] = 2s {Offset: 900, Size: 100, DurationTicks: 2000}, // [900..999] = 2s {Offset: 1400, Size: 100, DurationTicks: 2000}, }, }, }, }}
func TestRangeOverlapInTrack(t *testing.T) { meta := fixtureMetafile() video := meta.Tracks["1"] audio := meta.Tracks["2"]
t.Run("exact segment", func(t *testing.T) { b, d := rangeOverlapInTrack(0, 399, video) require.Equal(t, int64(400), b) require.Equal(t, int64(2000), d, "full segment ⇒ full duration") }) t.Run("half a segment", func(t *testing.T) { // First 200 bytes of segment 0 = half the bytes, half the duration. b, d := rangeOverlapInTrack(0, 199, video) require.Equal(t, int64(200), b) require.Equal(t, int64(1000), d, "byte-proportional duration credit") }) t.Run("range spanning two video segments + gap", func(t *testing.T) { // [350..599]: last 50 of seg0 video + audio gap + first 100 of seg1 video. b, d := rangeOverlapInTrack(350, 599, video) require.Equal(t, int64(50+100), b) // 50/400 * 2000 + 100/400 * 2000 = 250 + 500 = 750ms. require.Equal(t, int64(750), d) }) t.Run("range hits only audio offsets", func(t *testing.T) { b, _ := rangeOverlapInTrack(400, 499, video) require.Equal(t, int64(0), b, "audio bytes don't credit the video track") b, d := rangeOverlapInTrack(400, 499, audio) require.Equal(t, int64(100), b) require.Equal(t, int64(2000), d) }) t.Run("whole blob fetch", func(t *testing.T) { // [0..1499] covers every segment in both tracks. b, d := rangeOverlapInTrack(0, 1499, video) require.Equal(t, int64(400*3), b) require.Equal(t, int64(2000*3), d) b, d = rangeOverlapInTrack(0, 1499, audio) require.Equal(t, int64(100*3), b) require.Equal(t, int64(2000*3), d) }) t.Run("empty range", func(t *testing.T) { b, d := rangeOverlapInTrack(100, 50, video) require.Equal(t, int64(0), b) require.Equal(t, int64(0), d) }) t.Run("zero timescale punts", func(t *testing.T) { bad := vod.MetafileTrack{Timescale: 0, Segments: []vod.MetafileSegment{{Offset: 0, Size: 100, DurationTicks: 1000}}} b, d := rangeOverlapInTrack(0, 99, bad) require.Equal(t, int64(0), b) require.Equal(t, int64(0), d) })}
func TestAggregateWindowTrackUsage(t *testing.T) { root := t.TempDir() now := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) w := newAggTestWriter(t, root, "did:web:node1", now) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go w.Run(ctx)
const ( videoA = "at://did:plc:alice/place.stream.video/v1" cidA = "bafyfixture" sid = "session1" )
w.Log(ctx, Event{Ts: now, Type: EventTypeManifestRequest, VideoURI: videoA, SID: sid}) // Three segment_requests: video-seg0, audio-seg0, video-seg1. w.Log(ctx, Event{Ts: now.Add(time.Second), Type: EventTypeSegmentRequest, CID: cidA, SID: sid, RangeStart: 0, RangeEnd: 399}) w.Log(ctx, Event{Ts: now.Add(2 * time.Second), Type: EventTypeSegmentRequest, CID: cidA, SID: sid, RangeStart: 400, RangeEnd: 499}) w.Log(ctx, Event{Ts: now.Add(3 * time.Second), Type: EventTypeSegmentRequest, CID: cidA, SID: sid, RangeStart: 500, RangeEnd: 899}) require.NoError(t, w.Close())
store, err := blob.NewFileStore(root) require.NoError(t, err) refs := fixtureTrackRefs() res, err := AggregateWindow(context.Background(), store, AggregateInput{ WindowStart: now.Add(-time.Minute), WindowEnd: now.Add(time.Minute), FetchMetafile: func(ctx context.Context, cid string) (*vod.Metafile, error) { require.Equal(t, cidA, cid) return fixtureMetafile(), nil }, FetchTrackRefs: func(ctx context.Context, cid string) (map[string]comatproto.RepoStrongRef, error) { require.Equal(t, cidA, cid) return refs, nil }, }) require.NoError(t, err) require.Len(t, res.VideoCounts, 1) require.Equal(t, int64(1), res.VideoCounts[0].Count) require.Equal(t, 1, res.MetafilesLoaded, "metafile cache: one fetch even though three segment_requests share the CID")
byURI := map[string]TrackUsage{} for _, t := range res.VideoCounts[0].Tracks { byURI[t.Track.Uri] = t } require.Equal(t, TrackUsage{Track: refs["1"], Bytes: 800, DurationMS: 4000}, byURI[refs["1"].Uri], "two video segments fully fetched") require.Equal(t, TrackUsage{Track: refs["2"], Bytes: 100, DurationMS: 2000}, byURI[refs["2"].Uri], "one audio segment fully fetched")}
func TestAggregateWindowVideoWithUsageButNoCount(t *testing.T) { // Bytes flowed for a sid that never qualified as a view (e.g. an // orphan blob fetch from a previously-manifested sid that gets // thresholded out). The objective totals still ship. root := t.TempDir() now := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) w := newAggTestWriter(t, root, "did:web:node1", now) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go w.Run(ctx)
const ( videoA = "at://did:plc:alice/place.stream.video/v1" cidA = "bafyfixture" ) w.Log(ctx, Event{Ts: now, Type: EventTypeManifestRequest, VideoURI: videoA, SID: "lurker"}) w.Log(ctx, Event{Ts: now.Add(time.Second), Type: EventTypeSegmentRequest, CID: cidA, SID: "lurker", RangeStart: 0, RangeEnd: 399}) require.NoError(t, w.Close())
store, err := blob.NewFileStore(root) require.NoError(t, err) refs := fixtureTrackRefs() res, err := AggregateWindow(context.Background(), store, AggregateInput{ WindowStart: now.Add(-time.Minute), WindowEnd: now.Add(time.Minute), ThresholdSegments: 5, // far above the one segment fetched FetchMetafile: func(ctx context.Context, cid string) (*vod.Metafile, error) { return fixtureMetafile(), nil }, FetchTrackRefs: func(ctx context.Context, cid string) (map[string]comatproto.RepoStrongRef, error) { return refs, nil }, }) require.NoError(t, err) require.Len(t, res.VideoCounts, 1) require.Equal(t, int64(0), res.VideoCounts[0].Count, "below threshold ⇒ no view") require.Len(t, res.VideoCounts[0].Tracks, 1, "but the bytes are still reported") require.Equal(t, refs["1"].Uri, res.VideoCounts[0].Tracks[0].Track.Uri) require.Equal(t, int64(400), res.VideoCounts[0].Tracks[0].Bytes)}
func TestAggregateWindowTidWithoutStrongRefIsDropped(t *testing.T) { // A track that has a metafile entry but no place.stream.media.track // record (e.g. mid-publish, or the record was deleted) doesn't get // a usage row — we can't reference it. The sid view still counts. root := t.TempDir() now := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) w := newAggTestWriter(t, root, "did:web:node1", now) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go w.Run(ctx)
const ( videoA = "at://did:plc:alice/place.stream.video/v1" cidA = "bafyfixture" ) w.Log(ctx, Event{Ts: now, Type: EventTypeManifestRequest, VideoURI: videoA, SID: "sid"}) // Range hits both video (track "1") and audio (track "2"), but the // resolver only knows about track "1". w.Log(ctx, Event{Ts: now.Add(time.Second), Type: EventTypeSegmentRequest, CID: cidA, SID: "sid", RangeStart: 0, RangeEnd: 499}) require.NoError(t, w.Close())
store, err := blob.NewFileStore(root) require.NoError(t, err) refs := fixtureTrackRefs() partial := map[string]comatproto.RepoStrongRef{"1": refs["1"]} res, err := AggregateWindow(context.Background(), store, AggregateInput{ WindowStart: now.Add(-time.Minute), WindowEnd: now.Add(time.Minute), FetchMetafile: func(ctx context.Context, cid string) (*vod.Metafile, error) { return fixtureMetafile(), nil }, FetchTrackRefs: func(ctx context.Context, cid string) (map[string]comatproto.RepoStrongRef, error) { return partial, nil }, }) require.NoError(t, err) require.Len(t, res.VideoCounts, 1) require.Equal(t, int64(1), res.VideoCounts[0].Count) require.Len(t, res.VideoCounts[0].Tracks, 1, "only the resolved track gets a row") require.Equal(t, refs["1"].Uri, res.VideoCounts[0].Tracks[0].Track.Uri)}
func TestAggregateWindowMissingMetafileDoesNotPoison(t *testing.T) { // A segment_request whose CID has no metafile still counts toward // the sid view (the threshold cares about request count, not byte // math) — it just contributes zero bytes/duration. root := t.TempDir() now := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) w := newAggTestWriter(t, root, "did:web:node1", now) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go w.Run(ctx)
const videoA = "at://did:plc:alice/place.stream.video/v1" w.Log(ctx, Event{Ts: now, Type: EventTypeManifestRequest, VideoURI: videoA, SID: "sid"}) w.Log(ctx, Event{Ts: now.Add(time.Second), Type: EventTypeSegmentRequest, CID: "bafymissing", SID: "sid", RangeStart: 0, RangeEnd: 100}) require.NoError(t, w.Close())
store, err := blob.NewFileStore(root) require.NoError(t, err) res, err := AggregateWindow(context.Background(), store, AggregateInput{ WindowStart: now.Add(-time.Minute), WindowEnd: now.Add(time.Minute), FetchMetafile: func(ctx context.Context, cid string) (*vod.Metafile, error) { return nil, nil // not found }, }) require.NoError(t, err) require.Len(t, res.VideoCounts, 1) require.Equal(t, int64(1), res.VideoCounts[0].Count) require.Empty(t, res.VideoCounts[0].Tracks, "no metafile ⇒ no per-track credit")}
// TestAggregateWindowFlatHeaderOffset: node-logged Ranges are absolute// within the blob, which stores the fragments behind a synthesized// flat-MP4 header; the metafile's offsets are fragment-relative. A// request for the first video segment therefore arrives shifted by// FlatHeaderSize and must still credit exactly that segment.func TestAggregateWindowFlatHeaderOffset(t *testing.T) { root := t.TempDir() now := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) w := newAggTestWriter(t, root, "did:web:node1", now) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go w.Run(ctx)
const header = 1000 w.Log(ctx, Event{Ts: now, Type: EventTypeManifestRequest, VideoURI: "at://did:plc:alice/place.stream.video/v1", SID: "s"}) w.Log(ctx, Event{Ts: now.Add(time.Second), Type: EventTypeSegmentRequest, CID: "bafyfixture", SID: "s", RangeStart: header + 0, RangeEnd: header + 399}) require.NoError(t, w.Close())
store, err := blob.NewFileStore(root) require.NoError(t, err) refs := fixtureTrackRefs() res, err := AggregateWindow(context.Background(), store, AggregateInput{ WindowStart: now.Add(-time.Minute), WindowEnd: now.Add(time.Minute), FetchMetafile: func(ctx context.Context, cid string) (*vod.Metafile, error) { m := fixtureMetafile() m.FlatHeaderSize = header return m, nil }, FetchTrackRefs: func(ctx context.Context, cid string) (map[string]comatproto.RepoStrongRef, error) { return refs, nil }, }) require.NoError(t, err) require.Len(t, res.VideoCounts, 1) require.Len(t, res.VideoCounts[0].Tracks, 1, "only the video track was touched") require.Equal(t, TrackUsage{Track: refs["1"], Bytes: 400, DurationMS: 2000}, res.VideoCounts[0].Tracks[0])}
func TestProrateInTrack(t *testing.T) { meta := fixtureMetafile() total := metafileBodyBytes(meta) require.Equal(t, int64(1500), total)
// A CDN-logged request for 300 bytes with no Range: video holds // 1200/1500 of the body, audio 300/1500. Each track gets that // share of the bytes and the matching share of its own duration. b, d := prorateInTrack(300, total, meta.Tracks["1"]) require.Equal(t, int64(240), b) require.Equal(t, int64(1200), d, "6000ms of video * 300/1500") b, d = prorateInTrack(300, total, meta.Tracks["2"]) require.Equal(t, int64(60), b) require.Equal(t, int64(1200), d)
// Bytes beyond the body (a whole-blob fetch that included the flat // header) are capped at the body. b, d = prorateInTrack(99999, total, meta.Tracks["1"]) require.Equal(t, int64(1200), b) require.Equal(t, int64(6000), d)
b, d = prorateInTrack(0, total, meta.Tracks["1"]) require.Zero(t, b) require.Zero(t, d)}
// TestAggregateWindowBytesSentEvents: CDN-ingested events (BytesSent,// no Range) count toward the sid view and are prorated, alongside a// node-logged Range event for the same session.func TestAggregateWindowBytesSentEvents(t *testing.T) { root := t.TempDir() now := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) w := newAggTestWriter(t, root, "did:web:node1", now) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go w.Run(ctx)
w.Log(ctx, Event{Ts: now, Type: EventTypeManifestRequest, VideoURI: "at://did:plc:alice/place.stream.video/v1", SID: "s"}) w.Log(ctx, Event{Ts: now.Add(time.Second), Type: EventTypeSegmentRequest, CID: "bafyfixture", SID: "s", BytesSent: 1500}) require.NoError(t, w.Close())
store, err := blob.NewFileStore(root) require.NoError(t, err) refs := fixtureTrackRefs() res, err := AggregateWindow(context.Background(), store, AggregateInput{ WindowStart: now.Add(-time.Minute), WindowEnd: now.Add(time.Minute), FetchMetafile: func(ctx context.Context, cid string) (*vod.Metafile, error) { return fixtureMetafile(), nil }, FetchTrackRefs: func(ctx context.Context, cid string) (map[string]comatproto.RepoStrongRef, error) { return refs, nil }, }) require.NoError(t, err) require.Len(t, res.VideoCounts, 1) require.Equal(t, int64(1), res.VideoCounts[0].Count) byURI := map[string]TrackUsage{} for _, tu := range res.VideoCounts[0].Tracks { byURI[tu.Track.Uri] = tu } require.Equal(t, TrackUsage{Track: refs["1"], Bytes: 1200, DurationMS: 6000}, byURI[refs["1"].Uri]) require.Equal(t, TrackUsage{Track: refs["2"], Bytes: 300, DurationMS: 6000}, byURI[refs["2"].Uri])}