diff --git a/pkg/api/api_internal.go b/pkg/api/api_internal.go index ec04cf59..1b92b862 100644 --- a/pkg/api/api_internal.go +++ b/pkg/api/api_internal.go @@ -524,7 +524,7 @@ func (a *StreamplaceAPI) InternalHandler(ctx context.Context) (http.Handler, err } after := time.Now().Add(-time.Duration(secs) * time.Second) w.Header().Set("Content-Type", "video/mp4") - err = media.ClipUser(ctx, a.LocalDB, a.CLI, user, w, nil, &after) + err = a.MediaManager.ClipUser(ctx, user, w, nil, &after) if err != nil { errors.WriteHTTPInternalServerError(w, "unable to clip user", err) return diff --git a/pkg/media/clip_user.go b/pkg/media/clip_user.go deleted file mode 100644 index 04a6d2d4..00000000 --- a/pkg/media/clip_user.go +++ /dev/null @@ -1,54 +0,0 @@ -package media - -import ( - "context" - "fmt" - "io" - "os" - "sort" - "time" - - "stream.place/streamplace/pkg/aqtime" - "stream.place/streamplace/pkg/config" - "stream.place/streamplace/pkg/localdb" - "stream.place/streamplace/pkg/muxl" -) - -// ClipUser concatenates a user's stored canonical .m4s segments over the given -// time window into a single flat MP4 written to writer (used for -// moderation/report clips). The segments are blindly concatenatable MUXL, so -// the clip is just their bytes wrapped in a synthesized ftyp+moov — no remux, -// and each segment's C2PA signature passes through verbatim. -func ClipUser(ctx context.Context, localDB localdb.LocalDB, cli *config.CLI, user string, writer io.Writer, before *time.Time, after *time.Time) error { - segments, err := localDB.LatestSegmentsForUser(user, -1, false, before, after) - if err != nil { - return fmt.Errorf("unable to get segments: %w", err) - } - if len(segments) == 0 { - return fmt.Errorf("no segments found") - } - // Oldest first, so the clip plays in order. - sort.Slice(segments, func(i, j int) bool { - return segments[i].StartTime.Before(segments[j].StartTime) - }) - readers := make([]io.Reader, 0, len(segments)) - for _, segment := range segments { - aqt := aqtime.FromTime(segment.StartTime) - fpath, err := cli.SegmentFilePath(user, fmt.Sprintf("%s.%s", aqt.FileSafeString(), "m4s")) - if err != nil { - return fmt.Errorf("unable to get segment file path: %w", err) - } - fd, err := os.Open(fpath) - if err != nil { - return fmt.Errorf("unable to open segment file: %w", err) - } - defer fd.Close() - readers = append(readers, fd) - } - // muxl wrap unwraps every concatenated segment, aggregates their catalogs, - // and synthesizes one flat MP4 over the lot. - if err := muxl.RunMuxlWrap(ctx, io.MultiReader(readers...), "flat", writer); err != nil { - return fmt.Errorf("unable to clip segments: %w", err) - } - return nil -} diff --git a/pkg/media/media.go b/pkg/media/media.go index 7d8a1025..10c73843 100644 --- a/pkg/media/media.go +++ b/pkg/media/media.go @@ -40,9 +40,15 @@ const SegmentsDir = "segments" const StreamplaceMetadata = "cawg.metadata" type MediaManager struct { - cli *config.CLI - liveWindows map[string]*livehls.Writer - liveWindowsMut sync.Mutex + cli *config.CLI + liveWindows map[string]*livehls.Writer + liveWindowsMut sync.Mutex + // modBuffers holds a short in-memory ring of each live user's most recent + // canonical segments, the source for moderation/report clips now that + // segments are no longer archived to disk. Keyed by repoDID. See + // moderation_buffer.go. + modBuffers map[string]*modBuffer + modBuffersMut sync.Mutex httpPipes map[string]io.Writer httpPipesMutex sync.Mutex newSegmentSubs []chan *NewSegmentNotification @@ -120,6 +126,7 @@ func MakeMediaManager(ctx context.Context, cli *config.CLI, signer crypto.Signer mm := &MediaManager{ cli: cli, liveWindows: map[string]*livehls.Writer{}, + modBuffers: map[string]*modBuffer{}, httpPipes: map[string]io.Writer{}, model: mod, bus: bus, diff --git a/pkg/media/moderation_buffer.go b/pkg/media/moderation_buffer.go new file mode 100644 index 00000000..acd328ef --- /dev/null +++ b/pkg/media/moderation_buffer.go @@ -0,0 +1,135 @@ +package media + +import ( + "bytes" + "context" + "fmt" + "io" + "sort" + "time" + + "stream.place/streamplace/pkg/muxl" +) + +// modClipWindow is how long a user's recent canonical segments are kept in the +// in-memory moderation buffer. Moderation/report clips grab the last ~60s (see +// the makeClip / /clip handlers), so this is that window plus margin. It mirrors +// the old on-disk moderationRetention (120s) — the point at which a reported +// stream's evidence is still available — but holds the bytes in RAM instead of +// on disk. A hard modClipMaxSegments cap bounds memory against pathologically +// short segments. +const ( + modClipWindow = 2 * time.Minute + modClipMaxSegments = 240 +) + +// modSegment is one canonical segment retained for moderation clipping. +type modSegment struct { + startTime time.Time // media start time (what clips filter by) + addedAt time.Time // wall-clock arrival, for age eviction + data []byte // bare canonical .m4s bytes +} + +// modBuffer is a per-user time-bounded ring of recent canonical segments. All +// access goes through MediaManager (guarded by modBuffersMut), so it needs no +// lock of its own. +type modBuffer struct { + segs []modSegment +} + +// feedModerationBuffer appends a canonical segment to the user's moderation +// buffer and evicts anything older than modClipWindow. Called for every +// validated segment — published or not, so moderation can clip pre-live content +// exactly as the on-disk archive used to allow. The bytes are copied since the +// caller reuses its buffer. +func (mm *MediaManager) feedModerationBuffer(did string, startTime time.Time, segment []byte) { + now := time.Now() + mm.modBuffersMut.Lock() + defer mm.modBuffersMut.Unlock() + b := mm.modBuffers[did] + if b == nil { + b = &modBuffer{} + mm.modBuffers[did] = b + } + b.segs = append(b.segs, modSegment{ + startTime: startTime, + addedAt: now, + data: append([]byte(nil), segment...), + }) + b.evict(now) +} + +// evict drops segments older than the retention window and enforces the hard +// count cap (keeping the newest). Caller holds modBuffersMut. +func (b *modBuffer) evict(now time.Time) { + cutoff := now.Add(-modClipWindow) + drop := 0 + for drop < len(b.segs) && b.segs[drop].addedAt.Before(cutoff) { + drop++ + } + if drop > 0 { + b.segs = append(b.segs[:0:0], b.segs[drop:]...) + } + if len(b.segs) > modClipMaxSegments { + b.segs = append(b.segs[:0:0], b.segs[len(b.segs)-modClipMaxSegments:]...) + } +} + +// moderationSegments returns copies of the user's retained segments whose media +// start time falls in [after, before] (either bound nil = open), oldest first. +// Stale buffers (all segments aged out — a stream that ended) are dropped from +// the map so they're freed. +func (mm *MediaManager) moderationSegments(user string, before, after *time.Time) [][]byte { + now := time.Now() + mm.modBuffersMut.Lock() + defer mm.modBuffersMut.Unlock() + b := mm.modBuffers[user] + if b == nil { + return nil + } + b.evict(now) + if len(b.segs) == 0 { + delete(mm.modBuffers, user) + return nil + } + picked := make([]modSegment, 0, len(b.segs)) + for _, s := range b.segs { + if after != nil && s.startTime.Before(*after) { + continue + } + if before != nil && s.startTime.After(*before) { + continue + } + picked = append(picked, s) + } + sort.Slice(picked, func(i, j int) bool { + return picked[i].startTime.Before(picked[j].startTime) + }) + out := make([][]byte, len(picked)) + for i, s := range picked { + out[i] = s.data + } + return out +} + +// ClipUser concatenates a user's recent canonical .m4s segments over the given +// time window into a single flat MP4 written to writer (moderation/report +// clips). Segments come from the in-memory moderation buffer — on-disk archival +// was removed — so the window is bounded by modClipWindow; a request for a +// longer span yields only what is still retained. The segments are blindly +// concatenatable MUXL, so the clip is their bytes wrapped in a synthesized +// ftyp+moov, and each segment's C2PA signature passes through verbatim. +func (mm *MediaManager) ClipUser(ctx context.Context, user string, writer io.Writer, before, after *time.Time) error { + segs := mm.moderationSegments(user, before, after) + if len(segs) == 0 { + return fmt.Errorf("no segments found") + } + readers := make([]io.Reader, len(segs)) + for i, data := range segs { + readers[i] = bytes.NewReader(data) + } + if err := muxl.RunMuxlWrap(ctx, io.MultiReader(readers...), "flat", writer); err != nil { + return fmt.Errorf("unable to clip segments: %w", err) + } + return nil +} diff --git a/pkg/media/moderation_buffer_test.go b/pkg/media/moderation_buffer_test.go new file mode 100644 index 00000000..e348f1a2 --- /dev/null +++ b/pkg/media/moderation_buffer_test.go @@ -0,0 +1,67 @@ +package media + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func newTestMM() *MediaManager { + return &MediaManager{modBuffers: map[string]*modBuffer{}} +} + +// TestModerationBufferWindow verifies fed segments come back oldest-first and +// that the [after, before] filter selects the right slice by media start time. +func TestModerationBufferWindow(t *testing.T) { + mm := newTestMM() + base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + // Feed 5 segments at t=base+0s..+8s (2s apart), out of insertion order. + for _, i := range []int{2, 0, 4, 1, 3} { + mm.feedModerationBuffer("did:x", base.Add(time.Duration(i)*2*time.Second), []byte(fmt.Sprintf("seg%d", i))) + } + + all := mm.moderationSegments("did:x", nil, nil) + require.Len(t, all, 5) + require.Equal(t, []byte("seg0"), all[0], "oldest first") + require.Equal(t, []byte("seg4"), all[4], "newest last") + + // after = base+3s should drop seg0 (0s) and seg1 (2s), keep seg2..seg4. + after := base.Add(3 * time.Second) + got := mm.moderationSegments("did:x", nil, &after) + require.Len(t, got, 3) + require.Equal(t, []byte("seg2"), got[0]) + + // before = base+3s should keep only seg0 (0s) and seg1 (2s). + before := base.Add(3 * time.Second) + got = mm.moderationSegments("did:x", &before, nil) + require.Len(t, got, 2) + require.Equal(t, []byte("seg1"), got[1]) +} + +// TestModerationBufferCountCap verifies the hard segment cap keeps the newest +// modClipMaxSegments entries and drops the oldest. +func TestModerationBufferCountCap(t *testing.T) { + mm := newTestMM() + base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + total := modClipMaxSegments + 50 + for i := 0; i < total; i++ { + mm.feedModerationBuffer("did:x", base.Add(time.Duration(i)*time.Millisecond), []byte(fmt.Sprintf("s%d", i))) + } + got := mm.moderationSegments("did:x", nil, nil) + require.Len(t, got, modClipMaxSegments) + // The very first (oldest) segment must have been evicted; the newest kept. + require.Equal(t, []byte(fmt.Sprintf("s%d", total-1)), got[len(got)-1]) + require.Equal(t, []byte(fmt.Sprintf("s%d", total-modClipMaxSegments)), got[0]) +} + +// TestClipUserNoSegments verifies ClipUser reports "no segments" (so the +// moderation report still goes through) when the buffer is empty for a user. +func TestClipUserNoSegments(t *testing.T) { + mm := newTestMM() + err := mm.ClipUser(context.Background(), "did:unknown", nil, nil, nil) + require.Error(t, err) + require.Contains(t, err.Error(), "no segments") +} diff --git a/pkg/media/validate.go b/pkg/media/validate.go index cc471004..3cdd7637 100644 --- a/pkg/media/validate.go +++ b/pkg/media/validate.go @@ -210,23 +210,11 @@ func (mm *MediaManager) streamerIsBanned(repoDID string) (bool, error) { // synchronously from ValidateMP4 or asynchronously from the stream transcoder, // so it takes its own (non-request) context in the latter case. func (mm *MediaManager) distributeSegment(ctx context.Context, vs *validatedSegment, seg []byte) error { - tracer := otel.Tracer("signer") meta := vs.meta - _, fileSpan := tracer.Start(ctx, "ValidateMP4.SegmentArchiveWrite", trace.WithAttributes( - attribute.Int("bytes", len(seg)), - )) - fd, err := mm.cli.SegmentFileCreate(vs.repoDID, meta.StartTime, "m4s") - if err != nil { - fileSpan.End() - return err - } - defer fd.Close() - if _, err := io.Copy(fd, bytes.NewReader(seg)); err != nil { - fileSpan.End() - return err - } - fileSpan.End() + // Retain the canonical segment in the in-memory moderation buffer (this + // replaces on-disk .m4s archival) so reports can clip the recent window. + mm.feedModerationBuffer(vs.repoDID, meta.StartTime.Time(), seg) // Fold the validated segment into the streamer's live-HLS window, off the // critical path. This runs for every segment — locally signed or replicated @@ -235,13 +223,14 @@ func (mm *MediaManager) distributeSegment(ctx context.Context, vs *validatedSegm // Only published segments are actually folded in (see feedLiveWindow). go mm.feedLiveWindow(context.WithoutCancel(ctx), vs.repoDID, seg, meta.Published) - // The on-disk segment is transient now: durable copies live in S3/VOD, and - // disk archival is only a short-lived scratch for moderation clips + the - // no-S3 fallback. Every segment gets an expiry capped at SegmentArchiveRetention - // (default 1h) so nothing lingers — a -1 ("indefinite archival") distribution - // policy no longer pins segments to disk forever. An explicit *shorter* policy - // expiry still wins. SegmentArchiveRetention <= 0 is the operator opt-out that - // keeps segments until manually cleaned. + // Segments are no longer written to disk, but a Segment DB row is still kept + // (dedup, /segment metadata, live playlists). delete_after governs when that + // row is reaped by the cleaner — and reaps any legacy .m4s left on disk by an + // older build. Every row gets an expiry capped at SegmentArchiveRetention + // (default 1h); a -1 ("indefinite archival") distribution policy no longer + // pins it forever, and an explicit *shorter* policy expiry still wins. + // SegmentArchiveRetention <= 0 is the operator opt-out (keep rows until + // manually cleaned). var deleteAfter *time.Time if mm.cli.SegmentArchiveRetention > 0 { expiry := time.Now().Add(mm.cli.SegmentArchiveRetention).UTC() diff --git a/pkg/spxrpc/com_atproto_moderation.go b/pkg/spxrpc/com_atproto_moderation.go index f017909f..5d315ecc 100644 --- a/pkg/spxrpc/com_atproto_moderation.go +++ b/pkg/spxrpc/com_atproto_moderation.go @@ -13,7 +13,6 @@ import ( "github.com/labstack/echo/v4" "github.com/streamplace/oatproxy/pkg/oatproxy" "stream.place/streamplace/pkg/config" - "stream.place/streamplace/pkg/localdb" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/media" ) @@ -76,7 +75,7 @@ func (s *Server) handleComAtprotoModerationCreateReport(ctx context.Context, bod return nil, echo.NewHTTPError(http.StatusBadRequest, "invalid subject") } - clipID, err := makeClip(ctx, s.cli, s.localDB, did) + clipID, err := makeClip(ctx, s.cli, s.mm, did) if err != nil { // we still want the report to go through! log.Error(ctx, "failed to make clip for report", "error", err) @@ -99,7 +98,7 @@ func (s *Server) handleComAtprotoModerationCreateReport(ctx context.Context, bod return &output, nil } -func makeClip(ctx context.Context, cli *config.CLI, localDB localdb.LocalDB, did string) (string, error) { +func makeClip(ctx context.Context, cli *config.CLI, mm *media.MediaManager, did string) (string, error) { after := time.Now().Add(-time.Duration(60) * time.Second) uu, err := uuid.NewV7() @@ -113,7 +112,7 @@ func makeClip(ctx context.Context, cli *config.CLI, localDB localdb.LocalDB, did } defer fd.Close() - err = media.ClipUser(ctx, localDB, cli, did, fd, nil, &after) + err = mm.ClipUser(ctx, did, fd, nil, &after) if err != nil { return "", echo.NewHTTPError(http.StatusInternalServerError, "failed to clip user") }