From 0e35cac8d3baf6523314615e470b6d71ad5fcde6 Mon Sep 17 00:00:00 2001 From: Elliot Shine Date: Tue, 13 Oct 2026 17:21:38 +1030 Subject: [PATCH] oceanmedia: create oceanmedia service This change adds the oceanmedia service, which generates HLS playlists for oceanmedia broadcasts. Authentication is provided through delegated JWTs that are generated by the referring service (ie oceanbench or ausoceantv) and supplied using either a url parameter (to enable maximum compatibility with players) or bearer authentication. It will be deployed on the ausoceanmedia project (oceanmedia was taken). --- cmd/oceanmedia/auth.go | 72 ++++++ cmd/oceanmedia/main.go | 180 +++++++++++++++ cmd/oceanmedia/playlist.go | 248 +++++++++++++++++++++ cmd/oceanmedia/playlist_test.go | 381 ++++++++++++++++++++++++++++++++ cmd/oceanmedia/server.go | 131 +++++++++++ oceanmedia.yaml | 20 ++ scripts/run_all_standalone.sh | 2 + 7 files changed, 1034 insertions(+) create mode 100644 cmd/oceanmedia/auth.go create mode 100644 cmd/oceanmedia/main.go create mode 100644 cmd/oceanmedia/playlist.go create mode 100644 cmd/oceanmedia/playlist_test.go create mode 100644 cmd/oceanmedia/server.go create mode 100644 oceanmedia.yaml diff --git a/cmd/oceanmedia/auth.go b/cmd/oceanmedia/auth.go new file mode 100644 index 00000000..0b4ab0f4 --- /dev/null +++ b/cmd/oceanmedia/auth.go @@ -0,0 +1,72 @@ +/* +DESCRIPTION + OceanMedia JWT authentication. + +AUTHORS + Elliot Shine + +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 main + +import ( + "errors" + "fmt" + + "github.com/ausocean/cloud/gauth" + "github.com/gofiber/fiber/v2" +) + +var ( + errMissingToken = errors.New("missing token") + errInvalidToken = errors.New("invalid token") + errUnauthorized = errors.New("unauthorized") +) + +// authenticate validates the request's JWT and ensures that it authorises +// access to the given broadcast. Tokens are minted by referring services +// (such as OceanBench and AusOceanTV) using the shared JWT secret and +// supplied to players as the "token" query parameter. A Bearer token in +// the Authorization header is also accepted for programmatic clients. +// +// Required claims are "sub" (the broadcast ID) and "exp". +func (s *server) authenticate(c *fiber.Ctx, broadcastID string) error { + token := c.Query("token") + if token == "" { + token = c.Get(fiber.HeaderAuthorization) + } + if token == "" { + return errMissingToken + } + + claims, err := gauth.GetClaims(token, s.jwtSecret) + if err != nil { + return fmt.Errorf("%w: %v", errInvalidToken, err) + } + if _, ok := claims["exp"]; !ok { + return fmt.Errorf("%w: missing exp claim", errInvalidToken) + } + sub, ok := claims["sub"].(string) + if !ok || sub == "" { + return fmt.Errorf("%w: missing sub claim", errInvalidToken) + } + if sub != broadcastID { + return fmt.Errorf("%w: token is not valid for broadcast %s", errUnauthorized, broadcastID) + } + return nil +} diff --git a/cmd/oceanmedia/main.go b/cmd/oceanmedia/main.go new file mode 100644 index 00000000..ecbe6879 --- /dev/null +++ b/cmd/oceanmedia/main.go @@ -0,0 +1,180 @@ +/* +DESCRIPTION + OceanMedia generates HLS playlists for OceanMedia streams. + +AUTHORS + Elliot Shine + +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/. +*/ + +// OceanMedia serves HLS playlists for OceanMedia broadcasts. A +// broadcast is made up of a BroadcastEvent and a series of MtsMediaV2 +// segments whose media is stored in an object store bucket. Playlists +// are generated on demand with pre-signed segment URLs so that players +// fetch media directly from the bucket. +// +// Playlists support full refreshes, LL-HLS blocking reloads +// (_HLS_msn), delta updates (_HLS_skip) and DVR of the whole stream. +package main + +import ( + _ "github.com/joho/godotenv/autoload" + + "context" + "flag" + "fmt" + "log" + "os" + rtdebug "runtime/debug" + "strconv" + "sync" + "time" + + "github.com/ausocean/cloud/datastore" + "github.com/ausocean/cloud/gauth" + "github.com/ausocean/cloud/model" + "github.com/ausocean/cloud/storage" + "github.com/gofiber/fiber/v2" +) + +const ( + projectID = "oceanmedia" + version = "v0.1.0" +) + +// Secret names within OCEANMEDIA_SECRETS. +const ( + secretJWT = "jwtSecret" // JWT signing secret shared with referring services. + secretCloudflareAccount = "cloudflareAccountID" // Cloudflare R2 account ID. + secretCloudflareAccess = "cloudflareAccessKey" // Cloudflare R2 access key ID. + secretCloudflareSecret = "cloudflareSecretKey" // Cloudflare R2 secret access key. +) + +var ( + setupMutex sync.Mutex + mediaStore datastore.Store + srv *server + debug bool + standalone bool + storePath string + commitHash string +) + +func init() { + if info, ok := rtdebug.ReadBuildInfo(); ok { + for _, setting := range info.Settings { + if setting.Key == "vcs.revision" { + commitHash = setting.Value + if len(commitHash) > 7 { + commitHash = commitHash[:7] + } + break + } + } + } + if commitHash == "" { + commitHash = os.Getenv("COMMIT_HASH") + } +} + +func main() { + defaultPort := 8085 + v := os.Getenv("PORT") + if v != "" { + i, err := strconv.Atoi(v) + if err == nil { + defaultPort = i + } + } + + var ( + host string + port int + signedURLExpiry time.Duration + ) + flag.BoolVar(&debug, "debug", false, "Run in debug mode.") + flag.BoolVar(&standalone, "standalone", false, "Run in standalone mode.") + flag.StringVar(&host, "host", "localhost", "Host we run on in standalone mode") + flag.IntVar(&port, "port", defaultPort, "Port we listen on in standalone mode") + flag.StringVar(&storePath, "filestore", "store", "File store path") + flag.DurationVar(&signedURLExpiry, "signedurlttl", 12*time.Hour, "Pre-signed segment URL time to live") + flag.Parse() + + // Perform one-time setup or bail. + setup(context.Background(), signedURLExpiry) + + app := fiber.New(fiber.Config{DisableStartupMessage: true}) + + app.Get("/", indexHandler) + app.Get("/health", indexHandler) + app.Get("/_ah/warmup", indexHandler) + + // Playlists are requested as /.m3u8. + app.Get("/:file", srv.handlePlaylist) + + log.Printf("Listening on %s:%d", host, port) + log.Fatal(app.Listen(fmt.Sprintf("%s:%d", host, port))) +} + +// indexHandler handles health and liveness requests. +func indexHandler(c *fiber.Ctx) error { + if commitHash != "" { + return c.SendString(projectID + " " + version + " (" + commitHash + ")") + } + return c.SendString(projectID + " " + version) +} + +// setup executes per-instance one-time warmup and is used to +// initialize the datastores, secrets and signer. +func setup(ctx context.Context, signedURLExpiry time.Duration) { + setupMutex.Lock() + defer setupMutex.Unlock() + + if srv != nil { + return + } + + var err error + _, mediaStore, err = model.SetupDatastore(standalone, false, storePath, ctx) + if err != nil { + log.Fatalf("could not setup datastores: %v", err) + } + + secrets, err := gauth.GetSecrets(ctx, projectID, nil) + if err != nil { + log.Fatalf("could not get secrets: %v", err) + } + + jwtSecret := []byte(secrets[secretJWT]) + if len(jwtSecret) == 0 { + log.Fatalf("could not get %s, cannot validate tokens", secretJWT) + } + if secrets[secretCloudflareAccount] == "" { + log.Fatalf("could not get %s, cannot sign segment URLs", secretCloudflareAccount) + } + if secrets[secretCloudflareAccess] == "" || secrets[secretCloudflareSecret] == "" { + log.Fatalf("could not get %s/%s, cannot sign segment URLs", secretCloudflareAccess, secretCloudflareSecret) + } + + // The provider's account ID is used to reject segment URIs belonging to + // another account. The bucket is taken from each segment's StorageURI. + provider := storage.NewCloudflare(secrets[secretCloudflareAccount], secrets[secretCloudflareAccess], secrets[secretCloudflareSecret], "") + + // Both BroadcastEvent and MtsMediaV2 live in the media (vidgrind) datastore. + srv = newServer(mediaStore, provider, jwtSecret, signedURLExpiry) +} diff --git a/cmd/oceanmedia/playlist.go b/cmd/oceanmedia/playlist.go new file mode 100644 index 00000000..9d6cbc5c --- /dev/null +++ b/cmd/oceanmedia/playlist.go @@ -0,0 +1,248 @@ +/* +DESCRIPTION + OceanMedia HLS playlist generation. + +AUTHORS + Elliot Shine + +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 main + +import ( + "context" + "errors" + "fmt" + "math" + "strings" + "time" + + "github.com/ausocean/cloud/datastore" + "github.com/ausocean/cloud/model" + "github.com/gofiber/fiber/v2" +) + +// playlistMIME is the MIME type used for HLS playlists. +const playlistMIME = "application/vnd.apple.mpegurl" + +// skipUntilTargets is the minimum value of EXT-X-SERVER-CONTROL +// CAN-SKIP-UNTIL expressed in target durations. +const skipUntilTargets = 6 + +// handlePlaylist serves an HLS media playlist for a broadcast. +// +// The broadcast ID is taken from the request path, e.g. +// GET /.m3u8?token=. +// +// LL-HLS query parameters: +// +// _HLS_msn= - block until media sequence number n is available. +// _HLS_skip=YES - return a delta playlist (EXT-X-SKIP). +// _HLS_part= - unsupported; whole segments are served. +func (s *server) handlePlaylist(c *fiber.Ctx) error { + file := c.Params("file") + if !strings.HasSuffix(file, ".m3u8") { + return fiber.ErrNotFound + } + broadcastID := strings.TrimSuffix(file, ".m3u8") + if broadcastID == "" { + return fiber.ErrNotFound + } + + if err := s.authenticate(c, broadcastID); err != nil { + status := fiber.StatusForbidden + if errors.Is(err, errMissingToken) || errors.Is(err, errInvalidToken) { + status = fiber.StatusUnauthorized + } + return c.Status(status).JSON(fiber.Map{"error": err.Error()}) + } + + ctx := c.UserContext() + + var query struct { + MSN *int `query:"_HLS_msn"` + Part string `query:"_HLS_part"` + Skip string `query:"_HLS_skip"` + } + if err := c.QueryParser(&query); err != nil { + return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "invalid query parameters: " + err.Error()}) + } + if query.Part != "" { + return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "_HLS_part is not supported"}) + } + hasMSN := query.MSN != nil + msn := 0 + if hasMSN { + msn = *query.MSN + if msn < 0 { + return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": "invalid _HLS_msn"}) + } + } + skip := strings.EqualFold(query.Skip, "YES") + + event, err := s.loadEvent(ctx, broadcastID) + if err != nil { + if errors.Is(err, datastore.ErrNoSuchEntity) { + return fiber.ErrNotFound + } + s.log("could not get broadcast %s: %v", broadcastID, err) + return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "could not get broadcast"}) + } + + segments, err := s.loadSegments(ctx, broadcastID) + if err != nil { + s.log("could not get segments for broadcast %s: %v", broadcastID, err) + return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "could not get segments"}) + } + + if hasMSN { + segments, err = s.blockForMSN(ctx, broadcastID, msn, segments, event.EndTime.IsZero()) + if err != nil { + if errors.Is(err, errMSNTooFarAhead) { + return c.Status(fiber.StatusBadRequest).JSON(fiber.Map{"error": err.Error()}) + } + s.log("blocking reload for broadcast %s failed: %v", broadcastID, err) + return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "could not get segments"}) + } + } + + playlist, err := s.renderPlaylist(ctx, event, segments, msn, hasMSN, skip) + if err != nil { + s.log("could not render playlist for broadcast %s: %v", broadcastID, err) + return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{"error": "could not render playlist"}) + } + + c.Set(fiber.HeaderContentType, playlistMIME) + c.Set(fiber.HeaderCacheControl, "no-cache") + c.Set("Access-Control-Allow-Origin", "*") + return c.SendString(playlist) +} + +// errMSNTooFarAhead is returned when a requested media sequence number +// is beyond the next expected segment. +var errMSNTooFarAhead = errors.New("requested media sequence number is too far ahead") + +// blockForMSN waits until the segment with the given media sequence +// number is available, or the block timeout elapses. It returns the +// (possibly refreshed) segments and an error if the requested MSN is +// too far ahead. A non-live broadcast will never receive new segments, +// so blocking is skipped for it. +func (s *server) blockForMSN(ctx context.Context, broadcastID string, msn int, segments []model.MtsMediaV2, live bool) ([]model.MtsMediaV2, error) { + if msn < 0 { + return nil, errMSNTooFarAhead + } + // The client may request at most the next segment after the last + // available one. + if msn > len(segments) { + return nil, errMSNTooFarAhead + } + if msn < len(segments) { + return segments, nil + } + if !live { + // The broadcast has ended and the segment does not exist. + return nil, errMSNTooFarAhead + } + + deadline := time.Now().Add(s.blockTimeout) + for msn >= len(segments) && time.Now().Before(deadline) { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(s.blockPoll): + } + var err error + segments, err = s.loadSegments(ctx, broadcastID) + if err != nil { + return nil, err + } + } + return segments, nil +} + +// renderPlaylist builds an HLS media playlist. msn/hasMSN and skip +// control delta updates. The whole stream is included so that viewers +// can seek within a live stream. +func (s *server) renderPlaylist(ctx context.Context, event *model.BroadcastEvent, segments []model.MtsMediaV2, msn int, hasMSN, skip bool) (string, error) { + live := event.EndTime.IsZero() + + // When a client requests a delta we omit the segments it already + // has, i.e. those before the requested media sequence number. + first := 0 + if hasMSN && skip && msn > 0 && msn <= len(segments) { + first = msn + } + + // The target duration is the longest segment duration, rounded up. + var ( + target int64 = 1 + totalSecs float64 + ) + for _, seg := range segments { + secs := float64(seg.Duration) / 1000 + totalSecs += secs + if d := int64(math.Ceil(secs)); d > target { + target = d + } + } + + // The server can skip as far back as the whole playlist. + canSkipUntil := int64(math.Ceil(totalSecs)) + if min := skipUntilTargets * target; canSkipUntil < min { + canSkipUntil = min + } + + var b strings.Builder + b.WriteString("#EXTM3U\n") + // Version 9 is required for the LL-HLS tags used below + // (EXT-X-SERVER-CONTROL and EXT-X-SKIP). + b.WriteString("#EXT-X-VERSION:9\n") + fmt.Fprintf(&b, "#EXT-X-TARGETDURATION:%d\n", target) + // The playlist always begins at media sequence number 0 (the start of + // the broadcast). For a delta playlist, EXT-X-SKIP indicates how many + // leading segments have been omitted, so the first listed segment has + // media sequence number "first". + b.WriteString("#EXT-X-MEDIA-SEQUENCE:0\n") + b.WriteString("#EXT-X-PLAYLIST-TYPE:EVENT\n") + fmt.Fprintf(&b, "#EXT-X-SERVER-CONTROL:CAN-BLOCK-RELOAD=YES,CAN-SKIP-UNTIL=%d\n", canSkipUntil) + if first > 0 { + fmt.Fprintf(&b, "#EXT-X-SKIP:SKIPPED-SEGMENTS:%d\n", first) + } + + for i := first; i < len(segments); i++ { + seg := segments[i] + + signed, err := s.signURL(ctx, seg.StorageURI) + if err != nil { + return "", fmt.Errorf("segment %d: %w", i, err) + } + + if i > 0 && seg.Discontinuity { + b.WriteString("#EXT-X-DISCONTINUITY\n") + } + fmt.Fprintf(&b, "#EXTINF:%.3f,\n", float64(seg.Duration)/1000) + b.WriteString(signed) + b.WriteString("\n") + } + + if !live { + b.WriteString("#EXT-X-ENDLIST\n") + } + + return b.String(), nil +} diff --git a/cmd/oceanmedia/playlist_test.go b/cmd/oceanmedia/playlist_test.go new file mode 100644 index 00000000..9ceb4c16 --- /dev/null +++ b/cmd/oceanmedia/playlist_test.go @@ -0,0 +1,381 @@ +/* +DESCRIPTION + Tests for OceanMedia playlist handling. + +AUTHORS + Elliot Shine + +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 main + +import ( + "context" + "fmt" + "io" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/ausocean/cloud/datastore" + "github.com/ausocean/cloud/gauth" + "github.com/ausocean/cloud/model" + "github.com/gofiber/fiber/v2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + testBroadcastID = "c7cfd21b-d69d-437f-969c-34cd4ac15bd5" + testSegmentURL = "https://acct.r2.cloudflarestorage.com/bucket/" + testBroadcastID + "/seg%d.ts" +) + +var testJWTSecret = []byte("test-jwt-secret") + +// fakeSigner returns deterministic signed URLs, rejecting invalid URIs. +type fakeSigner struct{} + +func (fakeSigner) SignURL(_ context.Context, rawURI string, _ time.Duration) (string, error) { + if !strings.HasPrefix(rawURI, "https://") { + return "", fmt.Errorf("invalid storage URI %q", rawURI) + } + return "https://signed.example.com/" + rawURI + "?sig=x", nil +} + +// countingSigner counts SignURL calls for cache tests. +type countingSigner struct { + mu sync.Mutex + calls int +} + +func (c *countingSigner) SignURL(_ context.Context, _ string, _ time.Duration) (string, error) { + c.mu.Lock() + defer c.mu.Unlock() + c.calls++ + return "https://signed.example.com/" + strconv.Itoa(c.calls), nil +} + +func (c *countingSigner) callCount() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.calls +} + +func testSegments(n int, discontinuityAt int) []model.MtsMediaV2 { + segs := make([]model.MtsMediaV2, n) + for i := range segs { + segs[i] = model.MtsMediaV2{ + MID: 16, + Timestamp: 1790596912 + int64(i*3), + Duration: 2500, + StorageURI: fmt.Sprintf(testSegmentURL, i), + BroadcastID: testBroadcastID, + Discontinuity: discontinuityAt != -1 && i == discontinuityAt, + Type: "video/h264", + } + } + return segs +} + +func testToken(t *testing.T, sub string, exp time.Time) string { + t.Helper() + tok, err := gauth.PutClaims(map[string]interface{}{ + "sub": sub, + "exp": exp.Unix(), + }, testJWTSecret) + require.NoError(t, err) + return tok +} + +func newTestServer() *server { + s := newServer(nil, fakeSigner{}, testJWTSecret, time.Hour) + s.log = func(string, ...any) {} + return s +} + +func TestServerSignURLCache(t *testing.T) { + cs := &countingSigner{} + s := newServer(nil, cs, testJWTSecret, time.Hour) + + now := time.Now() + s.now = func() time.Time { return now } + + u1, err := s.signURL(context.Background(), "https://example.com/bucket/key.ts") + require.NoError(t, err) + u2, err := s.signURL(context.Background(), "https://example.com/bucket/key.ts") + require.NoError(t, err) + assert.Equal(t, u1, u2) + assert.Equal(t, 1, cs.callCount()) + + // Past half the time to live the URL should be re-signed. + now = now.Add(31 * time.Minute) + u3, err := s.signURL(context.Background(), "https://example.com/bucket/key.ts") + require.NoError(t, err) + assert.NotEqual(t, u1, u3) + assert.Equal(t, 2, cs.callCount()) +} + +func TestRenderPlaylistVOD(t *testing.T) { + s := newTestServer() + ctx := context.Background() + event := &model.BroadcastEvent{ID: testBroadcastID, EndTime: time.Now()} + + pl, err := s.renderPlaylist(ctx, event, testSegments(3, -1), 0, false, false) + require.NoError(t, err) + + assert.Contains(t, pl, "#EXTM3U") + assert.Contains(t, pl, "#EXT-X-VERSION:9") + assert.Contains(t, pl, "#EXT-X-TARGETDURATION:3") + assert.Contains(t, pl, "#EXT-X-MEDIA-SEQUENCE:0") + assert.Contains(t, pl, "#EXT-X-PLAYLIST-TYPE:EVENT") + assert.Contains(t, pl, "#EXT-X-SERVER-CONTROL:CAN-BLOCK-RELOAD=YES") + assert.Contains(t, pl, "#EXT-X-ENDLIST") + assert.Equal(t, 3, strings.Count(pl, "#EXTINF:2.500,")) + assert.Equal(t, 3, strings.Count(pl, "https://signed.example.com/")) + assert.NotContains(t, pl, "#EXT-X-SKIP") + assert.NotContains(t, pl, "#EXT-X-DISCONTINUITY") +} + +func TestRenderPlaylistLive(t *testing.T) { + s := newTestServer() + event := &model.BroadcastEvent{ID: testBroadcastID} + + pl, err := s.renderPlaylist(context.Background(), event, testSegments(2, -1), 0, false, false) + require.NoError(t, err) + assert.NotContains(t, pl, "#EXT-X-ENDLIST") + assert.Contains(t, pl, "#EXT-X-PLAYLIST-TYPE:EVENT") +} + +func TestRenderPlaylistLiveEmpty(t *testing.T) { + s := newTestServer() + event := &model.BroadcastEvent{ID: testBroadcastID} + + pl, err := s.renderPlaylist(context.Background(), event, nil, 0, false, false) + require.NoError(t, err) + assert.Contains(t, pl, "#EXTM3U") + assert.Contains(t, pl, "#EXT-X-TARGETDURATION:1") + assert.NotContains(t, pl, "#EXTINF") + assert.NotContains(t, pl, "#EXT-X-ENDLIST") +} + +func TestRenderPlaylistDiscontinuity(t *testing.T) { + s := newTestServer() + event := &model.BroadcastEvent{ID: testBroadcastID, EndTime: time.Now()} + + pl, err := s.renderPlaylist(context.Background(), event, testSegments(3, 1), 0, false, false) + require.NoError(t, err) + + assert.Equal(t, 1, strings.Count(pl, "#EXT-X-DISCONTINUITY")) + disc := strings.Index(pl, "#EXT-X-DISCONTINUITY") + seg0 := strings.Index(pl, "seg0.ts") + seg1 := strings.Index(pl, "seg1.ts") + assert.Greater(t, disc, seg0) + assert.Less(t, disc, seg1) +} + +func TestRenderPlaylistDelta(t *testing.T) { + s := newTestServer() + event := &model.BroadcastEvent{ID: testBroadcastID} + + pl, err := s.renderPlaylist(context.Background(), event, testSegments(3, -1), 2, true, true) + require.NoError(t, err) + + assert.Contains(t, pl, "#EXT-X-MEDIA-SEQUENCE:0") + assert.Contains(t, pl, "#EXT-X-SKIP:SKIPPED-SEGMENTS:2") + assert.Equal(t, 1, strings.Count(pl, "#EXTINF:")) + assert.Contains(t, pl, "seg2.ts") + assert.NotContains(t, pl, "seg0.ts") + assert.NotContains(t, pl, "seg1.ts") +} + +func TestRenderPlaylistSignError(t *testing.T) { + s := newTestServer() + event := &model.BroadcastEvent{ID: testBroadcastID} + segs := testSegments(1, -1) + segs[0].StorageURI = "not a uri" + + _, err := s.renderPlaylist(context.Background(), event, segs, 0, false, false) + assert.Error(t, err) +} + +func newTestApp(s *server) *fiber.App { + app := fiber.New() + app.Get("/:file", s.handlePlaylist) + return app +} + +func doRequest(t *testing.T, app *fiber.App, url string) (int, string) { + t.Helper() + resp, err := app.Test(httptest.NewRequest("GET", url, nil)) + require.NoError(t, err) + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + return resp.StatusCode, string(body) +} + +func TestHandlePlaylistSuccess(t *testing.T) { + s := newTestServer() + s.loadEvent = func(context.Context, string) (*model.BroadcastEvent, error) { + return &model.BroadcastEvent{ID: testBroadcastID, EndTime: time.Now()}, nil + } + s.loadSegments = func(context.Context, string) ([]model.MtsMediaV2, error) { + return testSegments(2, -1), nil + } + app := newTestApp(s) + + token := testToken(t, testBroadcastID, time.Now().Add(time.Hour)) + status, body := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token) + + assert.Equal(t, fiber.StatusOK, status) + assert.Contains(t, body, "#EXTM3U") + assert.Contains(t, body, "#EXT-X-ENDLIST") +} + +func TestHandlePlaylistAuth(t *testing.T) { + s := newTestServer() + s.loadEvent = func(context.Context, string) (*model.BroadcastEvent, error) { + return &model.BroadcastEvent{ID: testBroadcastID}, nil + } + s.loadSegments = func(context.Context, string) ([]model.MtsMediaV2, error) { + return testSegments(1, -1), nil + } + app := newTestApp(s) + + t.Run("Missing", func(t *testing.T) { + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8") + assert.Equal(t, fiber.StatusUnauthorized, status) + }) + + t.Run("Expired", func(t *testing.T) { + token := testToken(t, testBroadcastID, time.Now().Add(-time.Hour)) + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token) + assert.Equal(t, fiber.StatusUnauthorized, status) + }) + + t.Run("WrongBroadcast", func(t *testing.T) { + token := testToken(t, "some-other-broadcast", time.Now().Add(time.Hour)) + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token) + assert.Equal(t, fiber.StatusForbidden, status) + }) + + t.Run("BearerHeader", func(t *testing.T) { + token := testToken(t, testBroadcastID, time.Now().Add(time.Hour)) + req := httptest.NewRequest("GET", "/"+testBroadcastID+".m3u8", nil) + req.Header.Set("Authorization", "Bearer "+token) + resp, err := app.Test(req) + require.NoError(t, err) + assert.Equal(t, fiber.StatusOK, resp.StatusCode) + }) +} + +func TestHandlePlaylistNotFound(t *testing.T) { + s := newTestServer() + s.loadEvent = func(context.Context, string) (*model.BroadcastEvent, error) { + return nil, fmt.Errorf("failed to get BroadcastEvent: %w", datastore.ErrNoSuchEntity) + } + app := newTestApp(s) + + token := testToken(t, testBroadcastID, time.Now().Add(time.Hour)) + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token) + assert.Equal(t, fiber.StatusNotFound, status) +} + +func TestHandlePlaylistBadRequest(t *testing.T) { + s := newTestServer() + s.loadEvent = func(context.Context, string) (*model.BroadcastEvent, error) { + return &model.BroadcastEvent{ID: testBroadcastID}, nil + } + s.loadSegments = func(context.Context, string) ([]model.MtsMediaV2, error) { + return testSegments(2, -1), nil + } + app := newTestApp(s) + token := testToken(t, testBroadcastID, time.Now().Add(time.Hour)) + + t.Run("InvalidMSN", func(t *testing.T) { + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token+"&_HLS_msn=abc") + assert.Equal(t, fiber.StatusBadRequest, status) + }) + + t.Run("PartUnsupported", func(t *testing.T) { + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token+"&_HLS_part=0") + assert.Equal(t, fiber.StatusBadRequest, status) + }) + + t.Run("TooFarAhead", func(t *testing.T) { + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token+"&_HLS_msn=5") + assert.Equal(t, fiber.StatusBadRequest, status) + }) + + t.Run("TooFarAheadEmpty", func(t *testing.T) { + s.loadSegments = func(context.Context, string) ([]model.MtsMediaV2, error) { return nil, nil } + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token+"&_HLS_msn=1") + assert.Equal(t, fiber.StatusBadRequest, status) + }) + + t.Run("EndedNoBlocking", func(t *testing.T) { + // An ended broadcast will never receive new segments, so requesting + // the next media sequence number must not block; it is an error. + s.loadSegments = func(context.Context, string) ([]model.MtsMediaV2, error) { + return testSegments(2, -1), nil + } + s.loadEvent = func(context.Context, string) (*model.BroadcastEvent, error) { + return &model.BroadcastEvent{ID: testBroadcastID, EndTime: time.Now()}, nil + } + s.blockTimeout = time.Hour // Would hang if blocking were attempted. + status, _ := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token+"&_HLS_msn=2") + assert.Equal(t, fiber.StatusBadRequest, status) + }) +} + +func TestHandlePlaylistBlockingReload(t *testing.T) { + s := newTestServer() + s.blockPoll = time.Millisecond + s.blockTimeout = 2 * time.Second + s.loadEvent = func(context.Context, string) (*model.BroadcastEvent, error) { + return &model.BroadcastEvent{ID: testBroadcastID}, nil + } + var calls int + s.loadSegments = func(context.Context, string) ([]model.MtsMediaV2, error) { + calls++ + n := 1 + if calls > 1 { + n = 2 + } + return testSegments(n, -1), nil + } + app := newTestApp(s) + + token := testToken(t, testBroadcastID, time.Now().Add(time.Hour)) + status, body := doRequest(t, app, "/"+testBroadcastID+".m3u8?token="+token+"&_HLS_msn=1") + + assert.Equal(t, fiber.StatusOK, status) + assert.GreaterOrEqual(t, calls, 2) + assert.Equal(t, 2, strings.Count(body, "#EXTINF:")) + assert.Contains(t, body, "seg1.ts") +} + +func TestHandlePlaylistNotM3U8(t *testing.T) { + s := newTestServer() + app := newTestApp(s) + status, _ := doRequest(t, app, "/"+testBroadcastID) + assert.Equal(t, fiber.StatusNotFound, status) +} diff --git a/cmd/oceanmedia/server.go b/cmd/oceanmedia/server.go new file mode 100644 index 00000000..c5c36788 --- /dev/null +++ b/cmd/oceanmedia/server.go @@ -0,0 +1,131 @@ +/* +DESCRIPTION + OceanMedia server and dependencies. + +AUTHORS + Elliot Shine + +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 main + +import ( + "context" + "log" + "sync" + "time" + + "github.com/ausocean/cloud/datastore" + "github.com/ausocean/cloud/model" + "github.com/ausocean/cloud/storage" +) + +const ( + // defaultBlockPoll is how often the datastore is polled while + // waiting for a segment during a blocking reload. + defaultBlockPoll = 500 * time.Millisecond + + // defaultBlockTimeout is how long a blocking reload will wait for a + // segment before returning the current playlist. + defaultBlockTimeout = 15 * time.Second +) + +// urlSigner signs object storage URIs. It is satisfied by storage.Provider. +type urlSigner interface { + SignURL(ctx context.Context, rawURI string, ttl time.Duration) (string, error) +} + +type cachedURL struct { + url string + expires time.Time +} + +// server holds the dependencies used by OceanMedia's HTTP handlers. +type server struct { + store datastore.Store + signer urlSigner + jwtSecret []byte + signedURLTTL time.Duration + + blockPoll time.Duration + blockTimeout time.Duration + + // now is a field so that time can be controlled in tests. + now func() time.Time + + // Signed URLs are cached until half their lifetime has elapsed, so a + // URL handed to a client always has at least half its lifetime left. + signMu sync.Mutex + signCache map[string]cachedURL + + // loadEvent and loadSegments are fields so that they can be + // overridden in tests. + loadEvent func(ctx context.Context, broadcastID string) (*model.BroadcastEvent, error) + loadSegments func(ctx context.Context, broadcastID string) ([]model.MtsMediaV2, error) + + log func(string, ...any) +} + +// newServer returns a new server with the supplied dependencies. +func newServer(store datastore.Store, signer urlSigner, jwtSecret []byte, signedURLTTL time.Duration) *server { + if signedURLTTL <= 0 { + signedURLTTL = storage.DefaultSignedURLExpiry + } + s := &server{ + store: store, + signer: signer, + jwtSecret: jwtSecret, + signedURLTTL: signedURLTTL, + blockPoll: defaultBlockPoll, + blockTimeout: defaultBlockTimeout, + now: time.Now, + signCache: make(map[string]cachedURL), + log: log.Printf, + } + s.loadEvent = func(ctx context.Context, broadcastID string) (*model.BroadcastEvent, error) { + return model.GetBroadcastEvent(ctx, s.store, broadcastID) + } + s.loadSegments = func(ctx context.Context, broadcastID string) ([]model.MtsMediaV2, error) { + return model.GetMtsMediaV2ByBroadcast(ctx, s.store, broadcastID, nil, 0) + } + return s +} + +// signURL returns a pre-signed URL for the given storage URI, caching the +// result until half the signed URL time to live has elapsed. +func (s *server) signURL(ctx context.Context, rawURI string) (string, error) { + now := s.now() + + s.signMu.Lock() + if c, ok := s.signCache[rawURI]; ok && now.Before(c.expires) { + s.signMu.Unlock() + return c.url, nil + } + s.signMu.Unlock() + + signed, err := s.signer.SignURL(ctx, rawURI, s.signedURLTTL) + if err != nil { + return "", err + } + + s.signMu.Lock() + s.signCache[rawURI] = cachedURL{url: signed, expires: now.Add(s.signedURLTTL / 2)} + s.signMu.Unlock() + + return signed, nil +} diff --git a/oceanmedia.yaml b/oceanmedia.yaml new file mode 100644 index 00000000..3b2d4525 --- /dev/null +++ b/oceanmedia.yaml @@ -0,0 +1,20 @@ +# gcloud app deploy --version V --project ausoceanmedia oceanmedia.yaml +runtime: go126 + +env_variables: + AUSOCEAN_CREDENTIALS: gs://ausocean/AusOcean-ef9cd79f81a8.json + VIDGRIND_CREDENTIALS: gs://ausocean/VideoGrinder-b0ad82abac05.json + OCEANMEDIA_SECRETS: gs://ausocean/OceanMedia-secrets.txt + +main: cmd/oceanmedia + +handlers: + - url: /.* + secure: always + script: auto + +automatic_scaling: + min_instances: 1 + +inbound_services: + - warmup diff --git a/scripts/run_all_standalone.sh b/scripts/run_all_standalone.sh index 8a9c8b41..ee4c03c1 100755 --- a/scripts/run_all_standalone.sh +++ b/scripts/run_all_standalone.sh @@ -24,6 +24,7 @@ # Set the paths to your services OCEANBENCH_PATH="../cmd/oceanbench" OCEANTV_PATH="../cmd/oceantv" +OCEANMEDIA_PATH="../cmd/oceanmedia" OCEANCRON_PATH="../cmd/oceancron" DATABLUE_PATH="../cmd/datablue" @@ -79,6 +80,7 @@ trap cleanup SIGINT SIGTERM run_service "OceanBench" "$OCEANBENCH_PATH" "go run . --standalone --tvurl $TV_URL --cronurl $CRON_URL" run_service "OceanTV" "$OCEANTV_PATH" "go run . --standalone --filestore $FILESTORE_PATH" run_service "OceanCron" "$OCEANCRON_PATH" "go run . --standalone --filestore $FILESTORE_PATH" +run_service "OceanMedia" "$OCEANMEDIA_PATH" "go run . --standalone --filestore $FILESTORE_PATH" run_service "DataBlue" "$DATABLUE_PATH" "go run . --standalone --filestore $FILESTORE_PATH" # Wait for all background processes.