From bb63343f8230a32c6fb18bdcafa1ac60ccce2af3 Mon Sep 17 00:00:00 2001 From: Elliot Shine Date: Tue, 20 Oct 2026 13:41:38 +1030 Subject: [PATCH] model: add ClipID field and change key, datablue: ignore duplicate clips This change adds a ClipID field to the MtsMediaV2 entity that uniquely identifies a clip. This is added to the key so that duplicate writes fail. The mts v2 handler now skips clips that have already been written to the datastore. --- cmd/datablue/mts_handler.go | 5 ++ cmd/datablue/mts_handler_test.go | 16 +++-- model/mtsmediav2.go | 17 ++--- model/mtsmediav2_test.go | 103 ++++++++++++++++++++++++++++++- 4 files changed, 126 insertions(+), 15 deletions(-) diff --git a/cmd/datablue/mts_handler.go b/cmd/datablue/mts_handler.go index d9c44b74..dcd28aa6 100644 --- a/cmd/datablue/mts_handler.go +++ b/cmd/datablue/mts_handler.go @@ -24,6 +24,7 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "io" "log" @@ -236,6 +237,10 @@ func mtsHandlerV2(w http.ResponseWriter, r *http.Request) { mtsMedia.Timestamp = timestamp _, err = model.CreateMtsMediaV2(ctx, mediaStore, mtsMedia) if err != nil { + if errors.Is(err, datastore.ErrEntityExists) { + log.Printf("clip %s already exists, skipping", mtsMedia.ClipID) + continue + } log.Printf("could not write MTS media: %v", err) resp["er"] = fmt.Sprintf("could not write MTS media: %v", err) break diff --git a/cmd/datablue/mts_handler_test.go b/cmd/datablue/mts_handler_test.go index a1cbbe99..4bf86581 100644 --- a/cmd/datablue/mts_handler_test.go +++ b/cmd/datablue/mts_handler_test.go @@ -61,6 +61,7 @@ const ( testAudioType = "audio/aac" testDuration = 3000 testPTS = 80000 + testClipID = "clip-success" ) func setupTestStore(t *testing.T) context.Context { @@ -121,6 +122,7 @@ func TestMTSHandlerV2_Success(t *testing.T) { ctx := setupTestStore(t) meta := model.MtsMediaV2{ + ClipID: testClipID, Duration: testDuration, StorageURI: testVideoStorageURI, BroadcastID: testBroadcastID, @@ -151,11 +153,12 @@ func TestMTSHandlerV2_Success(t *testing.T) { assert.Nil(t, resp["er"], "unexpected error in response: %v", resp["er"]) mid := model.ToMID(testDevMac1, "V0") - stored, err := model.GetMtsMediaV2(ctx, mediaStore, mid, testTimestamp) + stored, err := model.GetMtsMediaV2(ctx, mediaStore, mid, testClipID) assert.NoError(t, err) // Check the stored entity has the correct values assert.Equal(t, mid, stored.MID, "expected MID %d, got %d", mid, stored.MID) + assert.Equal(t, testClipID, stored.ClipID, "expected ClipID %s, got %s", testClipID, stored.ClipID) assert.Equal(t, testGeohash, stored.Geohash, "expected Geohash %s, got %s", testGeohash, stored.Geohash) assert.Equal(t, testTimestamp, stored.Timestamp, "expected Timestamp %d, got %d", testTimestamp, stored.Timestamp) assert.Equal(t, meta.Duration, stored.Duration, "expected Duration %d, got %d", meta.Duration, stored.Duration) @@ -173,6 +176,7 @@ func TestMTSHandlerV2_DeviceLocation(t *testing.T) { _ = setupTestStore(t) meta := model.MtsMediaV2{ + ClipID: "clip-location", Duration: 2000, StorageURI: "gs://bucket/loc.ts", } @@ -199,11 +203,13 @@ func TestMTSHandlerV2_MultiplePins(t *testing.T) { ctx := setupTestStore(t) metaV0 := model.MtsMediaV2{ + ClipID: "clip-video", Duration: testDuration, StorageURI: testVideoStorageURI, Type: testVideoType, } metaS0 := model.MtsMediaV2{ + ClipID: "clip-audio", Duration: testDuration, StorageURI: testAudioStorageURI, Type: testAudioType, @@ -233,7 +239,7 @@ func TestMTSHandlerV2_MultiplePins(t *testing.T) { // Check that the V0 clip was stored correctly midV0 := model.ToMID(testDevMac1, "V0") - storedV0, err := model.GetMtsMediaV2(ctx, mediaStore, midV0, testTimestamp) + storedV0, err := model.GetMtsMediaV2(ctx, mediaStore, midV0, "clip-video") assert.NoError(t, err) assert.Equal(t, metaV0.StorageURI, storedV0.StorageURI) assert.Equal(t, metaV0.Type, storedV0.Type) @@ -244,7 +250,7 @@ func TestMTSHandlerV2_MultiplePins(t *testing.T) { // Check that the S0 clip was stored correctly midS0 := model.ToMID(testDevMac1, "S0") - storedS0, err := model.GetMtsMediaV2(ctx, mediaStore, midS0, testTimestamp) + storedS0, err := model.GetMtsMediaV2(ctx, mediaStore, midS0, "clip-audio") assert.NoError(t, err) assert.Equal(t, metaS0.StorageURI, storedS0.StorageURI) assert.Equal(t, metaS0.Type, storedS0.Type) @@ -259,7 +265,7 @@ func TestMTSHandlerV2_MultiplePins(t *testing.T) { func TestMTSHandlerV2_DefaultTimestamp(t *testing.T) { _ = setupTestStore(t) - meta := model.MtsMediaV2{Duration: 1000} + meta := model.MtsMediaV2{ClipID: "clip-default-ts", Duration: 1000} bodyBytes, _ := json.Marshal(meta) reqURL := fmt.Sprintf("/mtsv2?ma=%s&dk=%s&V0=%d", testDevMac1, testDevDkeyStr1, len(bodyBytes)) @@ -367,7 +373,7 @@ func TestMTSHandlerV2_ShortBody(t *testing.T) { // Read body error breaks loop before clip unmarshal and entity creation mid := model.ToMID(testDevMac1, "V0") - _, err = model.GetMtsMediaV2(ctx, mediaStore, mid, testTimestamp) + _, err = model.GetMtsMediaV2(ctx, mediaStore, mid, "clip-short-body") assert.Error(t, err, "expected no MtsMediaV2 entity to be stored on short body read") } diff --git a/model/mtsmediav2.go b/model/mtsmediav2.go index 1f3ed67a..14885352 100644 --- a/model/mtsmediav2.go +++ b/model/mtsmediav2.go @@ -46,6 +46,7 @@ const maxMtsMediaV2 = 20000 type MtsMediaV2 struct { MID int64 `json:"mid"` // Media ID. + ClipID string `json:"clipID"` // Unique clip identifier (e.g. a UUID), used as the entity key. Geohash string `json:"geohash"` // Geohash, if any. Timestamp int64 `json:"timestamp"` // Timestamp (in seconds). Duration int64 `json:"duration"` // Duration of the clip (in milliseconds). @@ -64,13 +65,13 @@ func (mts *MtsMediaV2) Copy(dst datastore.Entity) (datastore.Entity, error) { } // mtsMediaV2Key returns a datastore key for a MtsMediaV2 entity. -func mtsMediaV2Key(store datastore.Store, mid int64, timestamp int64) *datastore.Key { - return store.NameKey(typeMtsMediaV2, fmt.Sprintf("%d.%d", mid, timestamp)) +func mtsMediaV2Key(store datastore.Store, mid int64, clipID string) *datastore.Key { + return store.NameKey(typeMtsMediaV2, fmt.Sprintf("%d.%s", mid, clipID)) } // CreateMtsMediaV2 creates a new MtsMediaV2 entity. func CreateMtsMediaV2(ctx context.Context, store datastore.Store, mts MtsMediaV2) (*MtsMediaV2, error) { - key := mtsMediaV2Key(store, mts.MID, mts.Timestamp) + key := mtsMediaV2Key(store, mts.MID, mts.ClipID) mts.Date = time.Now() err := store.Create(ctx, key, &mts) if err != nil { @@ -80,8 +81,8 @@ func CreateMtsMediaV2(ctx context.Context, store datastore.Store, mts MtsMediaV2 } // GetMtsMediaV2 retrieves a MtsMediaV2 entity from the datastore. -func GetMtsMediaV2(ctx context.Context, store datastore.Store, mid int64, timestamp int64) (*MtsMediaV2, error) { - key := mtsMediaV2Key(store, mid, timestamp) +func GetMtsMediaV2(ctx context.Context, store datastore.Store, mid int64, clipID string) (*MtsMediaV2, error) { + key := mtsMediaV2Key(store, mid, clipID) mts := &MtsMediaV2{} err := store.Get(ctx, key, mts) if err != nil { @@ -91,8 +92,8 @@ func GetMtsMediaV2(ctx context.Context, store datastore.Store, mid int64, timest } // DeleteMtsMediaV2 deletes a MtsMediaV2 entity from the datastore. -func DeleteMtsMediaV2(ctx context.Context, store datastore.Store, mid int64, timestamp int64) error { - key := mtsMediaV2Key(store, mid, timestamp) +func DeleteMtsMediaV2(ctx context.Context, store datastore.Store, mid int64, clipID string) error { + key := mtsMediaV2Key(store, mid, clipID) err := store.Delete(ctx, key) if err != nil { return fmt.Errorf("failed to delete MtsMediaV2: %w", err) @@ -115,7 +116,7 @@ func GetMtsMediaV2ByBroadcast(ctx context.Context, store datastore.Store, broadc } // NB: no key parts are declared because BroadcastID is an entity - // field, not part of the MtsMediaV2 key (which is "."). + // field, not part of the MtsMediaV2 key (which is "."). q := store.NewQuery(typeMtsMediaV2, false) err := q.FilterField("BroadcastID", "=", broadcastID) if err != nil { diff --git a/model/mtsmediav2_test.go b/model/mtsmediav2_test.go index 5e81ab0a..62915c87 100644 --- a/model/mtsmediav2_test.go +++ b/model/mtsmediav2_test.go @@ -26,6 +26,8 @@ package model import ( "context" + "errors" + "fmt" "testing" "time" @@ -37,24 +39,37 @@ const ( testBcv2BroadcastA = "broadcast-v2-a" testBcv2BroadcastB = "broadcast-v2-b" testBcv2MID = int64(16) + + testIdemMID = int64(101) + testIdemTimestamp = int64(1700000000) + testIdemClipID = "clip-idempotent" + testIdemStorageURI = "https://bucket.example/clip-idempotent.ts" + testIdemBroadcastID = "test-mtsmediav2-idempotent" + + testGetMID = int64(102) + testGetTimestamp = int64(1700000001) + testGetClipID = "clip-get" + testGetBroadcastID = "test-mtsmediav2-get" ) // newMtsMediaV2 creates a new MtsMediaV2 entity, registering cleanup with t. // Its Date (used for ordering) is set to now by CreateMtsMediaV2. func newMtsMediaV2(t *testing.T, ctx context.Context, store datastore.Store, broadcastID string, mid, ts, pts int64) { t.Helper() + clipID := fmt.Sprintf("clip-%d-%d", ts, pts) _, err := CreateMtsMediaV2(ctx, store, MtsMediaV2{ MID: mid, + ClipID: clipID, Timestamp: ts, PTS: pts, Duration: 2541, - StorageURI: "https://example.com/bucket/" + broadcastID + "/segment.ts", + StorageURI: fmt.Sprintf("https://example.com/bucket/%s/segment-%d.ts", broadcastID, ts), BroadcastID: broadcastID, Type: "video/h264", }) assert.NoError(t, err) t.Cleanup(func() { - _ = DeleteMtsMediaV2(ctx, store, mid, ts) + _ = DeleteMtsMediaV2(ctx, store, mid, clipID) }) } @@ -115,3 +130,87 @@ func TestGetMtsMediaV2ByBroadcast(t *testing.T) { }) } } + +// TestCreateMtsMediaV2Idempotent checks that creating an MtsMediaV2 entity for +// a clip that has already been stored (same MID and ClipID) returns the +// existing entity rather than creating a duplicate. +func TestCreateMtsMediaV2Idempotent(t *testing.T) { + for _, kind := range storeKinds { + t.Run(kind, func(t *testing.T) { + ctx, store := setupStore(t, kind) + + mid := testIdemMID + base := testIdemTimestamp + clipID := testIdemClipID + uri := testIdemStorageURI + broadcastID := testIdemBroadcastID + t.Cleanup(func() { + _ = DeleteMtsMediaV2(ctx, store, mid, clipID) + }) + + first, err := CreateMtsMediaV2(ctx, store, MtsMediaV2{ + MID: mid, + ClipID: clipID, + Timestamp: base, + StorageURI: uri, + BroadcastID: broadcastID, + Duration: 2000, + }) + assert.NoError(t, err) + assert.NotNil(t, first) + + // A retry of the same clip, with a different receipt timestamp, + // must not create a duplicate. + _, err = CreateMtsMediaV2(ctx, store, MtsMediaV2{ + MID: mid, + ClipID: clipID, + Timestamp: base + 1, + StorageURI: uri, + BroadcastID: broadcastID, + Duration: 2000, + }) + assert.Error(t, err) + + media, err := GetMtsMediaV2ByBroadcast(ctx, store, broadcastID, 0) + assert.NoError(t, err) + assert.Len(t, media, 1, "expected exactly one entity for the clip") + }) + } +} + +// TestGetMtsMediaV2 checks that an entity can be retrieved by its clip ID and +// that a missing clip returns ErrNoSuchEntity. +func TestGetMtsMediaV2(t *testing.T) { + for _, kind := range storeKinds { + t.Run(kind, func(t *testing.T) { + ctx, store := setupStore(t, kind) + + mid := testGetMID + base := testGetTimestamp + clipID := testGetClipID + t.Cleanup(func() { + _ = DeleteMtsMediaV2(ctx, store, mid, clipID) + }) + + want := MtsMediaV2{ + MID: mid, + ClipID: clipID, + Timestamp: base, + StorageURI: "https://bucket.example/clip-get.ts", + BroadcastID: testGetBroadcastID, + Duration: 2000, + } + _, err := CreateMtsMediaV2(ctx, store, want) + assert.NoError(t, err) + + got, err := GetMtsMediaV2(ctx, store, mid, clipID) + assert.NoError(t, err) + assert.Equal(t, want.ClipID, got.ClipID) + assert.Equal(t, want.MID, got.MID) + assert.Equal(t, want.StorageURI, got.StorageURI) + + _, err = GetMtsMediaV2(ctx, store, mid, "no-such-clip") + assert.True(t, errors.Is(err, datastore.ErrNoSuchEntity), "expected ErrNoSuchEntity, got %v", err) + }) + } +}