model: add ClipID field and change key, datablue: ignore duplicate clips

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

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")
 }
 

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 "<mid>.<timestamp>").
+	// field, not part of the MtsMediaV2 key (which is "<mid>.<clipID>").
 	q := store.NewQuery(typeMtsMediaV2, false)
 	err := q.FilterField("BroadcastID", "=", broadcastID)
 	if err != nil {

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)
+		})
+	}
+}