model: add GetMtsMediaV2ByBroadcast

ausocean_test_index.yaml

@@ -77,6 +77,11 @@ indexes:
   - name: MID
   - name: Timestamp
 
+- kind: MtsMediaV2
+  properties:
+  - name: BroadcastID
+  - name: Timestamp
+
 - kind: Text
   properties:
   - name: MID

model/mtsmediav2.go

@@ -26,7 +26,9 @@ package model
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"sort"
 	"time"
 
 	"github.com/ausocean/cloud/datastore"
@@ -36,6 +38,12 @@ const (
 	typeMtsMediaV2 = "MtsMediaV2" // MtsMediaV2 datastore type.
 )
 
+// maxMtsMediaV2 is the default maximum number of MtsMediaV2 entities
+// returned for a broadcast when no limit is supplied. It is sized to
+// accommodate a full day of OceanMedia segments (an 8 hour broadcast at
+// roughly 2.5 second segments is around 12,000 segments).
+const maxMtsMediaV2 = 20000
+
 type MtsMediaV2 struct {
 	MID           int64     `json:"mid"`           // Media ID.
 	Geohash       string    `json:"geohash"`       // Geohash, if any.
@@ -91,3 +99,55 @@ func DeleteMtsMediaV2(ctx context.Context, store datastore.Store, mid int64, tim
 	}
 	return nil
 }
+
+// GetMtsMediaV2ByBroadcast retrieves the MtsMediaV2 entities belonging
+// to a broadcast, optionally filtered by timestamp(s). One timestamp
+// represents a lower bound (inclusive) on the segment timestamp,
+// whereas two represents a time range. Results are ordered by segment
+// timestamp ascending.
+//
+// A limit of zero or less uses a default size to cover the whole
+// broadcast.
+func GetMtsMediaV2ByBroadcast(ctx context.Context, store datastore.Store, broadcastID string, ts []int64, limit int) ([]MtsMediaV2, error) {
+	if broadcastID == "" {
+		return nil, errors.New("empty broadcast ID")
+	}
+	if limit <= 0 {
+		limit = maxMtsMediaV2
+	}
+
+	// NB: no key parts are declared because BroadcastID is an entity
+	// field, not part of the MtsMediaV2 key (which is "<mid>.<timestamp>").
+	q := store.NewQuery(typeMtsMediaV2, false)
+	err := q.FilterField("BroadcastID", "=", broadcastID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to filter MtsMediaV2 by broadcast ID: %w", err)
+	}
+	if len(ts) > 0 {
+		err = q.FilterField("Timestamp", ">=", ts[0])
+		if err != nil {
+			return nil, fmt.Errorf("failed to filter MtsMediaV2 by timestamp: %w", err)
+		}
+	}
+	if len(ts) > 1 {
+		err = q.FilterField("Timestamp", "<", ts[1])
+		if err != nil {
+			return nil, fmt.Errorf("failed to filter MtsMediaV2 by timestamp: %w", err)
+		}
+	}
+	q.Order("Timestamp")
+	q.Limit(limit)
+
+	var media []MtsMediaV2
+	_, err = store.GetAll(ctx, q, &media)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get MtsMediaV2 for broadcast %s: %w", broadcastID, err)
+	}
+
+	// FileStore orders by key name, which is not guaranteed to match a
+	// segment's timestamp, so sort explicitly.
+	sort.Slice(media, func(i, j int) bool {
+		return media[i].Timestamp < media[j].Timestamp
+	})
+	return media, nil
+}

model/mtsmediav2_test.go

@@ -0,0 +1,119 @@
+/*
+DESCRIPTION
+  Tests for model/mtsmediav2.go.
+
+AUTHORS
+  Elliot Shine <elliot@ausocean.org>
+
+LICENSE
+  Copyright (C) 2026 the Australian Ocean Lab (AusOcean).
+
+  This is free software: you can redistribute it and/or modify it
+  under the terms of the GNU General Public License as published by
+  the Free Software Foundation, either version 3 of the License, or
+  (at your option) any later version.
+
+  This is distributed in the hope that it will be useful, but WITHOUT
+  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
+  or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public
+  License for more details.
+
+  You should have received a copy of the GNU General Public License in
+  gpl.txt. If not, see http://www.gnu.org/licenses/.
+*/
+
+package model
+
+import (
+	"context"
+	"testing"
+
+	"github.com/ausocean/cloud/datastore"
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	testBcv2BroadcastA = "broadcast-v2-a"
+	testBcv2BroadcastB = "broadcast-v2-b"
+	testBcv2MID        = int64(16)
+)
+
+// newMtsMediaV2 creates a new MtsMediaV2 entity with the given broadcast
+// ID and timestamp, registering cleanup with t.
+func newMtsMediaV2(t *testing.T, ctx context.Context, store datastore.Store, broadcastID string, mid, ts int64) {
+	t.Helper()
+	_, err := CreateMtsMediaV2(ctx, store, MtsMediaV2{
+		MID:         mid,
+		Timestamp:   ts,
+		Duration:    2541,
+		StorageURI:  "https://example.com/bucket/" + broadcastID + "/segment.ts",
+		BroadcastID: broadcastID,
+		Type:        "video/h264",
+	})
+	assert.NoError(t, err)
+	t.Cleanup(func() {
+		_ = DeleteMtsMediaV2(ctx, store, mid, ts)
+	})
+}
+
+func TestGetMtsMediaV2ByBroadcast(t *testing.T) {
+	for _, kind := range storeKinds {
+		t.Run(kind, func(t *testing.T) {
+			ctx, store := setupStore(t, kind)
+
+			for _, ts := range []int64{1000, 1005, 1010, 1015} {
+				newMtsMediaV2(t, ctx, store, testBcv2BroadcastA, testBcv2MID, ts)
+			}
+			newMtsMediaV2(t, ctx, store, testBcv2BroadcastB, testBcv2MID, 2000)
+
+			timestamps := func(media []MtsMediaV2) []int64 {
+				out := make([]int64, len(media))
+				for i, m := range media {
+					out[i] = m.Timestamp
+				}
+				return out
+			}
+
+			t.Run("AllOrdered", func(t *testing.T) {
+				got, err := GetMtsMediaV2ByBroadcast(ctx, store, testBcv2BroadcastA, nil, 0)
+				assert.NoError(t, err)
+				assert.Equal(t, []int64{1000, 1005, 1010, 1015}, timestamps(got))
+			})
+
+			t.Run("BroadcastFilter", func(t *testing.T) {
+				got, err := GetMtsMediaV2ByBroadcast(ctx, store, testBcv2BroadcastB, nil, 0)
+				assert.NoError(t, err)
+				assert.Equal(t, []int64{2000}, timestamps(got))
+			})
+
+			t.Run("FromTimestamp", func(t *testing.T) {
+				got, err := GetMtsMediaV2ByBroadcast(ctx, store, testBcv2BroadcastA, []int64{1005}, 0)
+				assert.NoError(t, err)
+				assert.Equal(t, []int64{1005, 1010, 1015}, timestamps(got))
+			})
+
+			t.Run("TimestampRange", func(t *testing.T) {
+				got, err := GetMtsMediaV2ByBroadcast(ctx, store, testBcv2BroadcastA, []int64{1005, 1015}, 0)
+				assert.NoError(t, err)
+				assert.Equal(t, []int64{1005, 1010}, timestamps(got))
+			})
+
+			t.Run("Limit", func(t *testing.T) {
+				got, err := GetMtsMediaV2ByBroadcast(ctx, store, testBcv2BroadcastA, nil, 2)
+				assert.NoError(t, err)
+				assert.Equal(t, []int64{1000, 1005}, timestamps(got))
+			})
+
+			t.Run("NoResults", func(t *testing.T) {
+				got, err := GetMtsMediaV2ByBroadcast(ctx, store, "no-such-broadcast", nil, 0)
+				assert.NoError(t, err)
+				assert.Empty(t, got)
+			})
+
+			t.Run("EmptyBroadcastID", func(t *testing.T) {
+				_, err := GetMtsMediaV2ByBroadcast(ctx, store, "", nil, 0)
+				assert.Error(t, err)
+			})
+		})
+	}
+}

vidgrind_index.yaml

@@ -5,6 +5,11 @@ indexes:
   - name: MID
   - name: Timestamp
 
+- kind: MtsMediaV2
+  properties:
+  - name: BroadcastID
+  - name: Timestamp
+
 - kind: Text
   properties:
   - name: MID