diff --git a/pkg/api/api.go b/pkg/api/api.go index 66510c6f..23b466d6 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -47,6 +47,7 @@ import ( "stream.place/streamplace/pkg/statedb" "stream.place/streamplace/pkg/streamplace" "stream.place/streamplace/pkg/upload" + "stream.place/streamplace/pkg/viewlog" metrics "github.com/slok/go-http-metrics/metrics/prometheus" "github.com/slok/go-http-metrics/middleware" @@ -68,6 +69,7 @@ type StreamplaceAPI struct { MediaSigner media.MediaSigner UploadManager *upload.Manager PlaybackStore blob.Store + ViewLog *viewlog.Writer XRPCServer *spxrpc.Server // not thread-safe yet Aliases map[string]string @@ -99,7 +101,7 @@ type WebsocketTracker struct { mu sync.RWMutex } -func MakeStreamplaceAPI(cli *config.CLI, mod model.Model, statefulDB *statedb.StatefulDB, noter notifications.FirebaseNotifier, mm *media.MediaManager, ms media.MediaSigner, bus *bus.Bus, atsync *atproto.ATProtoSynchronizer, d *director.Director, op *oatproxy.OATProxy, ldb localdb.LocalDB, um *upload.Manager, playbackStore blob.Store) (*StreamplaceAPI, error) { +func MakeStreamplaceAPI(cli *config.CLI, mod model.Model, statefulDB *statedb.StatefulDB, noter notifications.FirebaseNotifier, mm *media.MediaManager, ms media.MediaSigner, bus *bus.Bus, atsync *atproto.ATProtoSynchronizer, d *director.Director, op *oatproxy.OATProxy, ldb localdb.LocalDB, um *upload.Manager, playbackStore blob.Store, viewLog *viewlog.Writer) (*StreamplaceAPI, error) { updater, err := PrepareUpdater(cli) if err != nil { return nil, err @@ -113,6 +115,7 @@ func MakeStreamplaceAPI(cli *config.CLI, mod model.Model, statefulDB *statedb.St MediaSigner: ms, UploadManager: um, PlaybackStore: playbackStore, + ViewLog: viewLog, Aliases: map[string]string{}, Bus: bus, ATSync: atsync, @@ -161,7 +164,7 @@ func (a *StreamplaceAPI) Handler(ctx context.Context) (http.Handler, error) { Recorder: metrics.NewRecorder(metrics.Config{}), }) var xrpc http.Handler - xrpc, err := spxrpc.NewServer(ctx, a.CLI, a.Model, a.StatefulDB, a.op, mdlw, a.ATSync, a.Bus, a.LocalDB, a.MediaManager, a.UploadManager, a.PlaybackStore, a.Aliases) + xrpc, err := spxrpc.NewServer(ctx, a.CLI, a.Model, a.StatefulDB, a.op, mdlw, a.ATSync, a.Bus, a.LocalDB, a.MediaManager, a.UploadManager, a.PlaybackStore, a.ViewLog, a.Aliases) if err != nil { return nil, err } diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index 307ebec2..bb2143c3 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -43,6 +43,7 @@ import ( "stream.place/streamplace/pkg/statedb" "stream.place/streamplace/pkg/storage" "stream.place/streamplace/pkg/upload" + "stream.place/streamplace/pkg/viewlog" "stream.place/streamplace/pkg/vod" "github.com/aws/aws-sdk-go-v2/aws" @@ -358,6 +359,21 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu if err != nil { return fmt.Errorf("make vod store: %w", err) } + viewLog, err := makeViewLog(ctx, cli, vodStore, ldb) + if err != nil { + return fmt.Errorf("make view log: %w", err) + } + if viewLog != nil { + group.Go(func() error { + viewLog.Run(ctx) + return nil + }) + defer func() { + if err := viewLog.Close(); err != nil { + log.Error(ctx, "view log close", "error", err) + } + }() + } state.SetVODProcessor(func(ctx context.Context, t statedb.VODProcessTask) (string, error) { // Labeler enforcement: an account banned after starting an // upload (but before processing) doesn't get a video published. @@ -380,7 +396,7 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu Location: t.Location, }) }) - a, err := api.MakeStreamplaceAPI(cli, mod, state, noter, mm, ms, b, atsync, d, op, ldb, um, vodStore) + a, err := api.MakeStreamplaceAPI(cli, mod, state, noter, mm, ms, b, atsync, d, op, ldb, um, vodStore, viewLog) if err != nil { return err } @@ -625,6 +641,34 @@ func makeVODStore(ctx context.Context, cli *config.CLI) (blob.Store, error) { return blob.NewFileStore(root) } +// makeViewLog returns the configured view-event log writer, or nil if +// the operator has disabled it (--view-log-flush-interval=0) or there's +// no place to write logs to (no VOD store). The writer reuses the VOD +// blob.Store under a `view-logs//` prefix; if the operator +// runs S3-backed VOD, view logs land in the same bucket alongside the +// content blobs and pick up the bucket's lifecycle policy for free. +func makeViewLog(ctx context.Context, cli *config.CLI, vodStore blob.Store, ldb localdb.LocalDB) (*viewlog.Writer, error) { + if cli.ViewLogFlushInterval <= 0 { + log.Log(ctx, "view log: disabled (view-log-flush-interval=0)") + return nil, nil + } + if vodStore == nil { + log.Log(ctx, "view log: no VOD store wired; skipping") + return nil, nil + } + w, err := viewlog.NewWriter(viewlog.Config{ + Store: vodStore, + NodeDID: cli.ServerDID(), + FlushAfter: cli.ViewLogFlushInterval, + Salts: viewlog.NewSaltManager(ldb), + }) + if err != nil { + return nil, err + } + log.Log(ctx, "view log: enabled", "flush_interval", cli.ViewLogFlushInterval, "node_did", cli.ServerDID()) + return w, nil +} + var ErrCaughtSignal = errors.New("caught signal") func handleSignals(ctx context.Context) error { diff --git a/pkg/config/config.go b/pkg/config/config.go index 4d919bd3..9966f465 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -164,6 +164,7 @@ type CLI struct { GamesAPIClientKey string GamesAPIClientSecret string BetaInviteDID string + ViewLogFlushInterval time.Duration } // ContentFilters represents the content filtering configuration @@ -937,6 +938,13 @@ func (cli *CLI) NewCommand(name string) *urfavecli.Command { Destination: &cli.BetaInviteDID, Sources: urfavecli.EnvVars("SP_BETA_INVITE_DID"), }, + &urfavecli.DurationFlag{ + Name: "view-log-flush-interval", + Usage: "How often the view-log writer rotates its buffer to the VOD blob store. Set to 0 to disable view-event logging entirely (no view counts will be available downstream). Files land at view-logs//.jsonl.gz alongside the VOD content blobs.", + Value: 5 * time.Minute, + Destination: &cli.ViewLogFlushInterval, + Sources: urfavecli.EnvVars("SP_VIEW_LOG_FLUSH_INTERVAL"), + }, &urfavecli.BoolFlag{ Name: "legacy-segmentation", Usage: "switch back from MUXL to legacy segmentation in case streams have problems (shouldn't need!)", diff --git a/pkg/localdb/localdb.go b/pkg/localdb/localdb.go index 64ee5b15..b85ff5b8 100644 --- a/pkg/localdb/localdb.go +++ b/pkg/localdb/localdb.go @@ -26,6 +26,9 @@ type LocalDB interface { DeleteSegment(ctx context.Context, id string) error StartSegmentCleaner(ctx context.Context) error SegmentCleaner(ctx context.Context) error + GetViewLogSalt(date string) ([]byte, error) + PutViewLogSalt(date string, salt []byte) error + DeleteViewLogSaltsBefore(date string) error } type LocalDatabase struct { @@ -71,6 +74,7 @@ func MakeDB(dbURL string) (LocalDB, error) { for _, model := range []any{ Segment{}, Thumbnail{}, + ViewLogSalt{}, } { err = db.AutoMigrate(model) if err != nil { diff --git a/pkg/localdb/view_log_salt.go b/pkg/localdb/view_log_salt.go new file mode 100644 index 00000000..355a3417 --- /dev/null +++ b/pkg/localdb/view_log_salt.go @@ -0,0 +1,38 @@ +package localdb + +import ( + "errors" + "fmt" + + "gorm.io/gorm" +) + +// ViewLogSalt is the per-UTC-day HMAC salt used to anonymize IP +// addresses in the view-log pipeline. Same date returns the same salt, +// so two requests from the same IP within a day share a hash; across +// days the hash changes because the salt rotates. Cleaning up old rows +// makes historical logs unrecoverable. +type ViewLogSalt struct { + Date string `gorm:"primaryKey;column:date"` + Salt []byte `gorm:"column:salt"` +} + +func (m *LocalDatabase) GetViewLogSalt(date string) ([]byte, error) { + var row ViewLogSalt + err := m.DB.Where("date = ?", date).First(&row).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("get view log salt: %w", err) + } + return row.Salt, nil +} + +func (m *LocalDatabase) PutViewLogSalt(date string, salt []byte) error { + return m.DB.Save(&ViewLogSalt{Date: date, Salt: salt}).Error +} + +func (m *LocalDatabase) DeleteViewLogSaltsBefore(date string) error { + return m.DB.Where("date < ?", date).Delete(&ViewLogSalt{}).Error +} diff --git a/pkg/spxrpc/place_stream_playback_getvideo.go b/pkg/spxrpc/place_stream_playback_getvideo.go index 8bdafae4..31539ce0 100644 --- a/pkg/spxrpc/place_stream_playback_getvideo.go +++ b/pkg/spxrpc/place_stream_playback_getvideo.go @@ -142,9 +142,26 @@ func (s *Server) HandleGetVideoBlob(c echo.Context) error { return echo.NewHTTPError(http.StatusInternalServerError, err.Error()) } defer r.Close() + rangeStart, rangeEnd := parseRangeForLog(c.Request().Header.Get("Range"), r.Size()) + s.logSegmentRequest(c, cid, did, c.QueryParam("sid"), rangeStart, rangeEnd) return serveBlobRange(c, r, "video/mp4") } +// parseRangeForLog is the lenient counterpart of parseSingleRange used +// only for the view-log: returns zero values when the header is absent +// or malformed (logging shouldn't 416 the request the way serving +// does). End is inclusive when present, matching RFC 7233. +func parseRangeForLog(header string, size int64) (int64, int64) { + if header == "" { + return 0, 0 + } + start, end, err := parseSingleRange(header, size) + if err != nil { + return 0, 0 + } + return start, end +} + // serveBlobRange writes a blob.Reader to the echo response, honoring // HTTP Range. We only support single ranges (`bytes=N-M`, `bytes=N-`, // `bytes=-N`); multi-range requests are rejected with 416. Cache @@ -317,6 +334,7 @@ func (s *Server) HandleGetVideoPlaylist(c echo.Context) error { effectiveStartMS, effectiveEndMS := composeClipBounds(resolved.clipStartMS, resolved.clipEndMS, startMS, endMS) var body string + kind := "master" if track == "" { // Master playlist's sub-playlist URLs carry the *unmodified* // query-param start/end (clip-local). Each per-track follow-up @@ -328,7 +346,9 @@ func (s *Server) HandleGetVideoPlaylist(c echo.Context) error { if err != nil { return err } + kind = "media" } + s.logManifestRequest(c, uri, sid, track, kind) c.Response().Header().Set("Content-Type", "application/vnd.apple.mpegurl") c.Response().Header().Set("Cache-Control", "public, max-age=60") diff --git a/pkg/spxrpc/spxrpc.go b/pkg/spxrpc/spxrpc.go index 26d3c0d7..2a9b36af 100644 --- a/pkg/spxrpc/spxrpc.go +++ b/pkg/spxrpc/spxrpc.go @@ -25,6 +25,7 @@ import ( "stream.place/streamplace/pkg/model" "stream.place/streamplace/pkg/statedb" "stream.place/streamplace/pkg/upload" + "stream.place/streamplace/pkg/viewlog" ) type Server struct { @@ -45,10 +46,14 @@ type Server struct { // segments live. Matches the blob.Store the VOD processor writes // into (vod.BlobsPrefix + .{mp4,json}). playbackStore blob.Store - aliases map[string]string + // viewLog records playback request events (manifest + segment + // fetches) for later view-count aggregation. Optional; nil when + // --view-log-flush-interval is 0 or no playback store is wired. + viewLog *viewlog.Writer + aliases map[string]string } -func NewServer(ctx context.Context, cli *config.CLI, model model.Model, statefulDB *statedb.StatefulDB, op *oatproxy.OATProxy, mdlw middleware.Middleware, atsync *atproto.ATProtoSynchronizer, bus *bus.Bus, ldb localdb.LocalDB, mm *media.MediaManager, um *upload.Manager, playbackStore blob.Store, aliases map[string]string) (*Server, error) { +func NewServer(ctx context.Context, cli *config.CLI, model model.Model, statefulDB *statedb.StatefulDB, op *oatproxy.OATProxy, mdlw middleware.Middleware, atsync *atproto.ATProtoSynchronizer, bus *bus.Bus, ldb localdb.LocalDB, mm *media.MediaManager, um *upload.Manager, playbackStore blob.Store, viewLog *viewlog.Writer, aliases map[string]string) (*Server, error) { e := echo.New() s := &Server{ e: e, @@ -65,6 +70,7 @@ func NewServer(ctx context.Context, cli *config.CLI, model model.Model, stateful mm: mm, uploadManager: um, playbackStore: playbackStore, + viewLog: viewLog, aliases: aliases, } e.Use(s.ErrorHandlingMiddleware()) diff --git a/pkg/spxrpc/viewlog_hooks.go b/pkg/spxrpc/viewlog_hooks.go new file mode 100644 index 00000000..d37c91a8 --- /dev/null +++ b/pkg/spxrpc/viewlog_hooks.go @@ -0,0 +1,64 @@ +package spxrpc + +import ( + "time" + + "github.com/labstack/echo/v4" + + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/viewlog" +) + +// logManifestRequest captures a playlist fetch (master or media) for +// later view-count aggregation. No-op when --view-log-flush-interval=0 +// has left s.viewLog nil. +func (s *Server) logManifestRequest(c echo.Context, uri, sid, track, kind string) { + if s.viewLog == nil { + return + } + ctx := c.Request().Context() + now := time.Now().UTC() + ipHash, err := s.viewLog.Salts().HashIP(c.RealIP(), now) + if err != nil { + // Drop the hash but still log the event — better a partial + // record than a missed view. + log.Error(ctx, "viewlog: hash IP", "error", err) + ipHash = "" + } + s.viewLog.Log(ctx, viewlog.Event{ + Ts: now, + Type: viewlog.EventTypeManifestRequest, + VideoURI: uri, + SID: sid, + IPHash: ipHash, + ManifestKind: kind, + Track: track, + }) +} + +// logSegmentRequest captures a content-addressed blob fetch. The +// aggregator joins CID → place.stream.video.AT-URI via the MediaTrack +// index (one row per (track, blob)); we log raw CID + owner here so +// the handler doesn't pay a DB roundtrip on the hot path. +func (s *Server) logSegmentRequest(c echo.Context, cid, ownerDID, sid string, rangeStart, rangeEnd int64) { + if s.viewLog == nil { + return + } + ctx := c.Request().Context() + now := time.Now().UTC() + ipHash, err := s.viewLog.Salts().HashIP(c.RealIP(), now) + if err != nil { + log.Error(ctx, "viewlog: hash IP", "error", err) + ipHash = "" + } + s.viewLog.Log(ctx, viewlog.Event{ + Ts: now, + Type: viewlog.EventTypeSegmentRequest, + SID: sid, + IPHash: ipHash, + CID: cid, + OwnerDID: ownerDID, + RangeStart: rangeStart, + RangeEnd: rangeEnd, + }) +} diff --git a/pkg/viewlog/event.go b/pkg/viewlog/event.go new file mode 100644 index 00000000..8bb3b259 --- /dev/null +++ b/pkg/viewlog/event.go @@ -0,0 +1,72 @@ +// Package viewlog captures playback request events (manifest + +// segment fetches) to a blob.Store as gzipped JSONL files, so a later +// aggregation pass can derive view counts per place.stream.video. +// +// Goals at this layer: cheap to call from request handlers, non- +// blocking, durable enough that a periodic flush survives a crash +// with bounded data loss (the in-memory buffer between flushes). +// +// Privacy: IPs are HMAC'd with a daily-rotated salt held in localdb. +// Same IP collides within a UTC day (useful for dedup); the salt +// changes at midnight so cross-day correlation requires the original +// salt, which the operator can prune to enforce a retention horizon. +package viewlog + +import "time" + +// Event type tags. +const ( + EventTypeManifestRequest = "manifest_request" + EventTypeSegmentRequest = "segment_request" +) + +// Manifest kinds populated on ManifestRequest events. +const ( + ManifestKindMaster = "master" + ManifestKindMedia = "media" +) + +// Event is one line in the JSONL output. The Type field selects which +// of the per-type field groups are meaningful — empty fields are +// omitted on the wire to keep files small. +type Event struct { + // Ts is the server's UTC clock at request time. + Ts time.Time `json:"ts"` + // Type is one of EventType*. Aggregators dispatch on this. + Type string `json:"type"` + // SID is the playback session id (atproto TID) the player carries + // across manifest + segment fetches. Generated by the server on + // the master playlist response and threaded through sub-playlist + // + segment URLs from there. + SID string `json:"sid,omitempty"` + // IPHash is hex(HMAC-SHA256(daily_salt, ip)). Same IP, same UTC + // day, same hash; salt rotation at midnight breaks correlation + // across days. Empty when no IP could be resolved. + IPHash string `json:"ip_hash,omitempty"` + + // --- manifest_request --- + // VideoURI is the place.stream.video AT-URI the manifest is for. + VideoURI string `json:"video_uri,omitempty"` + // ManifestKind is "master" or "media". + ManifestKind string `json:"manifest_kind,omitempty"` + // Track is the muxlTrack id when ManifestKind == "media"; empty + // for master playlists (those span every track). + Track string `json:"track,omitempty"` + + // --- segment_request --- + // CID is the content-addressed blob requested. For a playback + // segment this is the MUXL container blob shared across the + // video's tracks; for init segments it's the per-track init + // blob. Aggregators join CID → place.stream.video via the + // MediaTrack index. + CID string `json:"cid,omitempty"` + // OwnerDID is the repo DID the segment URL claims as the owner. + // Lets the aggregator attribute traffic to a creator without a + // CID lookup; verified upstream at the playback handler. + OwnerDID string `json:"owner_did,omitempty"` + // RangeStart / RangeEnd capture the HTTP Range bytes, when the + // request asked for one. End is inclusive (matches RFC 7233). + // Zero values indicate "whole blob" / "open-ended" range. + RangeStart int64 `json:"range_start,omitempty"` + RangeEnd int64 `json:"range_end,omitempty"` +} diff --git a/pkg/viewlog/salt.go b/pkg/viewlog/salt.go new file mode 100644 index 00000000..da000d06 --- /dev/null +++ b/pkg/viewlog/salt.go @@ -0,0 +1,86 @@ +package viewlog + +import ( + "crypto/hmac" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "fmt" + "sync" + "time" +) + +// SaltStorage persists per-UTC-day HMAC salts. The localdb satisfies +// it; tests can substitute an in-memory implementation. Concrete +// methods are expected to handle "not present" by returning a nil +// salt + nil error so the manager can mint one on first use. +type SaltStorage interface { + GetViewLogSalt(date string) ([]byte, error) + PutViewLogSalt(date string, salt []byte) error +} + +// SaltDateFormat is the canonical UTC date key used by the storage +// layer. Exposed so callers (e.g. retention prune) can format their +// own dates consistently. +const SaltDateFormat = "2006-01-02" + +// SaltManager caches the per-day salt in memory; the first request of +// a UTC day mints a fresh 32-byte salt and persists it, every later +// request returns the cached value. +// +// Safe for concurrent use. +type SaltManager struct { + storage SaltStorage + + mu sync.Mutex + cache map[string][]byte +} + +func NewSaltManager(storage SaltStorage) *SaltManager { + return &SaltManager{ + storage: storage, + cache: make(map[string][]byte), + } +} + +// Salt returns the salt for t's UTC date, minting + persisting one on +// the first call for a given day. +func (m *SaltManager) Salt(t time.Time) ([]byte, error) { + date := t.UTC().Format(SaltDateFormat) + m.mu.Lock() + defer m.mu.Unlock() + if salt, ok := m.cache[date]; ok { + return salt, nil + } + salt, err := m.storage.GetViewLogSalt(date) + if err != nil { + return nil, fmt.Errorf("read view-log salt for %s: %w", date, err) + } + if salt == nil { + salt = make([]byte, 32) + if _, err := rand.Read(salt); err != nil { + return nil, fmt.Errorf("generate view-log salt: %w", err) + } + if err := m.storage.PutViewLogSalt(date, salt); err != nil { + return nil, fmt.Errorf("persist view-log salt for %s: %w", date, err) + } + } + m.cache[date] = salt + return salt, nil +} + +// HashIP returns hex(HMAC-SHA256(daily_salt, ip)). Empty ip returns +// an empty string so callers can pass through whatever c.RealIP() +// gives them without a branch. +func (m *SaltManager) HashIP(ip string, t time.Time) (string, error) { + if ip == "" { + return "", nil + } + salt, err := m.Salt(t) + if err != nil { + return "", err + } + mac := hmac.New(sha256.New, salt) + mac.Write([]byte(ip)) + return hex.EncodeToString(mac.Sum(nil)), nil +} diff --git a/pkg/viewlog/salt_test.go b/pkg/viewlog/salt_test.go new file mode 100644 index 00000000..40a09f98 --- /dev/null +++ b/pkg/viewlog/salt_test.go @@ -0,0 +1,122 @@ +package viewlog + +import ( + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// memSaltStorage is the in-memory SaltStorage used by tests so they +// don't need a real localdb. +type memSaltStorage struct { + mu sync.Mutex + salts map[string][]byte + getErr error + putErr error + getHits int + putHits int +} + +func newMemSaltStorage() *memSaltStorage { + return &memSaltStorage{salts: make(map[string][]byte)} +} + +func (m *memSaltStorage) GetViewLogSalt(date string) ([]byte, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.getHits++ + if m.getErr != nil { + return nil, m.getErr + } + return m.salts[date], nil +} + +func (m *memSaltStorage) PutViewLogSalt(date string, salt []byte) error { + m.mu.Lock() + defer m.mu.Unlock() + m.putHits++ + if m.putErr != nil { + return m.putErr + } + m.salts[date] = append([]byte(nil), salt...) + return nil +} + +func TestSaltManagerMintsAndCachesPerDay(t *testing.T) { + storage := newMemSaltStorage() + mgr := NewSaltManager(storage) + + day1 := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) + salt1, err := mgr.Salt(day1) + require.NoError(t, err) + require.Len(t, salt1, 32) + require.Equal(t, 1, storage.getHits, "first call reads storage") + require.Equal(t, 1, storage.putHits, "first call persists the minted salt") + + // Second call same day, same UTC day → cached, no storage hits. + salt1b, err := mgr.Salt(day1.Add(3 * time.Hour)) + require.NoError(t, err) + require.Equal(t, salt1, salt1b) + require.Equal(t, 1, storage.getHits, "subsequent same-day calls don't re-read storage") + require.Equal(t, 1, storage.putHits, "subsequent same-day calls don't re-write storage") + + // Different day → new salt, new storage hit. + day2 := day1.Add(24 * time.Hour) + salt2, err := mgr.Salt(day2) + require.NoError(t, err) + require.NotEqual(t, salt1, salt2, "salt must change across days") + require.Equal(t, 2, storage.getHits) + require.Equal(t, 2, storage.putHits) +} + +func TestSaltManagerReusesPersistedSalt(t *testing.T) { + // Simulates a process restart: pre-populated storage, new manager + // instance, same day → same salt comes back. + storage := newMemSaltStorage() + preset := []byte("00112233445566778899aabbccddeeff") + require.NoError(t, storage.PutViewLogSalt("2026-05-17", preset)) + storage.putHits = 0 // reset to count subsequent writes only + + mgr := NewSaltManager(storage) + day := time.Date(2026, 5, 17, 0, 0, 0, 0, time.UTC) + got, err := mgr.Salt(day) + require.NoError(t, err) + require.Equal(t, preset, got) + require.Equal(t, 0, storage.putHits, "no write when storage already has the salt") +} + +func TestHashIPSameIPSameDayCollides(t *testing.T) { + mgr := NewSaltManager(newMemSaltStorage()) + now := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) + + h1, err := mgr.HashIP("203.0.113.7", now) + require.NoError(t, err) + h2, err := mgr.HashIP("203.0.113.7", now.Add(2*time.Hour)) + require.NoError(t, err) + require.Equal(t, h1, h2, "same IP within UTC day collides") + + other, err := mgr.HashIP("198.51.100.42", now) + require.NoError(t, err) + require.NotEqual(t, h1, other, "different IPs differ within a day") +} + +func TestHashIPDifferentDayDecorrelates(t *testing.T) { + mgr := NewSaltManager(newMemSaltStorage()) + day1 := time.Date(2026, 5, 17, 23, 30, 0, 0, time.UTC) + day2 := day1.Add(2 * time.Hour) // crosses UTC midnight + + h1, err := mgr.HashIP("203.0.113.7", day1) + require.NoError(t, err) + h2, err := mgr.HashIP("203.0.113.7", day2) + require.NoError(t, err) + require.NotEqual(t, h1, h2, "same IP across days must produce different hashes") +} + +func TestHashIPEmptyIPYieldsEmpty(t *testing.T) { + mgr := NewSaltManager(newMemSaltStorage()) + got, err := mgr.HashIP("", time.Now()) + require.NoError(t, err) + require.Empty(t, got, "no IP → no hash; callers can pass c.RealIP() straight through") +} diff --git a/pkg/viewlog/writer.go b/pkg/viewlog/writer.go new file mode 100644 index 00000000..b0d2f74d --- /dev/null +++ b/pkg/viewlog/writer.go @@ -0,0 +1,214 @@ +package viewlog + +import ( + "bytes" + "compress/gzip" + "context" + "encoding/json" + "fmt" + "sync" + "time" + + "stream.place/streamplace/pkg/blob" + "stream.place/streamplace/pkg/log" +) + +// Writer buffers Event values gzipped in memory and periodically +// uploads the buffer as a single object to its backing blob.Store. +// Each upload's key is the writer's keyPrefix + the buffer's opened-at +// timestamp + ".jsonl.gz", so files for a window are easy to list + +// time-order at aggregation time. +// +// Log is fast (one mutex, one JSON encode into a gzip writer); the +// upload happens off the request goroutine. A Run loop drives time- +// based flushes; size-based flushes are nudged via flushReq. +type Writer struct { + store blob.Store + keyPrefix string + flushAfter time.Duration + maxBytes int + salts *SaltManager + now func() time.Time + + mu sync.Mutex + buf *bytes.Buffer + gz *gzip.Writer + openedAt time.Time + eventCount int + unflushedBytes int // pre-compression; gzip's internal buffer hides post-compression size until Close + + flushReq chan struct{} + closeCh chan struct{} + doneCh chan struct{} +} + +// Config bundles writer settings. NodeDID identifies the writer in the +// key layout: `view-logs//.jsonl.gz`. A CDN ETL would +// use the same shape with a CDN-side identifier. +type Config struct { + Store blob.Store + NodeDID string + FlushAfter time.Duration + MaxBytes int + Salts *SaltManager + // Now is an optional clock override for tests. Defaults to time.Now. + Now func() time.Time +} + +func NewWriter(cfg Config) (*Writer, error) { + if cfg.Store == nil { + return nil, fmt.Errorf("viewlog: Store is required") + } + if cfg.NodeDID == "" { + return nil, fmt.Errorf("viewlog: NodeDID is required") + } + if cfg.FlushAfter <= 0 { + cfg.FlushAfter = 5 * time.Minute + } + if cfg.MaxBytes <= 0 { + cfg.MaxBytes = 10 * 1024 * 1024 + } + if cfg.Now == nil { + cfg.Now = func() time.Time { return time.Now().UTC() } + } + w := &Writer{ + store: cfg.Store, + keyPrefix: fmt.Sprintf("view-logs/%s/", cfg.NodeDID), + flushAfter: cfg.FlushAfter, + maxBytes: cfg.MaxBytes, + salts: cfg.Salts, + now: cfg.Now, + flushReq: make(chan struct{}, 1), + closeCh: make(chan struct{}), + doneCh: make(chan struct{}), + } + w.reset() + return w, nil +} + +// Salts returns the writer's SaltManager. Handlers use it to hash IPs +// before logging. +func (w *Writer) Salts() *SaltManager { return w.salts } + +// reset opens a fresh in-memory gzip buffer. Called from the writer +// loop while the mutex is held. +func (w *Writer) reset() { + w.buf = new(bytes.Buffer) + w.gz = gzip.NewWriter(w.buf) + w.openedAt = w.now() + w.eventCount = 0 + w.unflushedBytes = 0 +} + +// Log encodes ev into the current buffer and nudges a flush if the +// uncompressed byte count crossed maxBytes. We track pre-compression +// bytes because gzip's internal buffer makes w.buf.Len() useless as a +// size signal between flushes. Errors here are logged and swallowed +// — a playback handler can't usefully recover from a view-log write +// failure, and propagating would change the playback contract. +func (w *Writer) Log(ctx context.Context, ev Event) { + line, err := json.Marshal(&ev) + if err != nil { + log.Error(ctx, "viewlog: encode event", "error", err, "type", ev.Type) + return + } + line = append(line, '\n') + + w.mu.Lock() + _, werr := w.gz.Write(line) + if werr == nil { + w.unflushedBytes += len(line) + w.eventCount++ + } + overSize := w.unflushedBytes >= w.maxBytes + w.mu.Unlock() + if werr != nil { + log.Error(ctx, "viewlog: gzip write", "error", werr, "type", ev.Type) + return + } + if overSize { + select { + case w.flushReq <- struct{}{}: + default: + // A flush is already queued; nothing to do. + } + } +} + +// Run blocks until ctx is cancelled or Close is called, periodically +// (every flushAfter / 2) checking for an open buffer with events and +// flushing it. Size-based flushes arrive on flushReq from Log. +func (w *Writer) Run(ctx context.Context) { + defer close(w.doneCh) + tick := time.NewTicker(w.flushAfter) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + w.flushNow(context.Background(), "context-done") + return + case <-w.closeCh: + w.flushNow(context.Background(), "close") + return + case <-tick.C: + w.flushNow(ctx, "tick") + case <-w.flushReq: + w.flushNow(ctx, "size") + } + } +} + +// Close signals Run to flush and exit, then waits for it. +func (w *Writer) Close() error { + select { + case <-w.closeCh: + // Already closed. + default: + close(w.closeCh) + } + <-w.doneCh + return nil +} + +// flushNow uploads whatever's currently buffered as a single +// .jsonl.gz blob; a no-op if the buffer is empty. Reason is recorded +// on the log line for ops visibility (size vs tick vs close). +func (w *Writer) flushNow(ctx context.Context, reason string) { + w.mu.Lock() + if w.eventCount == 0 { + w.mu.Unlock() + return + } + oldBuf := w.buf + oldGz := w.gz + oldOpenedAt := w.openedAt + oldCount := w.eventCount + w.reset() + w.mu.Unlock() + + if err := oldGz.Close(); err != nil { + log.Error(ctx, "viewlog: gzip close", "error", err) + return + } + key := w.keyPrefix + oldOpenedAt.UTC().Format("2006-01-02T15-04-05.000Z") + ".jsonl.gz" + writer, err := w.store.NewWriter(ctx, key, "application/gzip") + if err != nil { + log.Error(ctx, "viewlog: open store writer", "error", err, "key", key) + return + } + if _, err := writer.Write(oldBuf.Bytes()); err != nil { + log.Error(ctx, "viewlog: write blob", "error", err, "key", key) + _ = writer.Close() + return + } + if err := writer.Complete(); err != nil { + log.Error(ctx, "viewlog: complete blob", "error", err, "key", key) + return + } + log.Debug(ctx, "viewlog flushed", + "key", key, + "events", oldCount, + "bytes", oldBuf.Len(), + "reason", reason, + ) +} diff --git a/pkg/viewlog/writer_test.go b/pkg/viewlog/writer_test.go new file mode 100644 index 00000000..8d3ff2a7 --- /dev/null +++ b/pkg/viewlog/writer_test.go @@ -0,0 +1,202 @@ +package viewlog + +import ( + "bufio" + "compress/gzip" + "context" + "encoding/json" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "stream.place/streamplace/pkg/blob" +) + +// readAllJSONL gunzips and JSON-decodes every line in the given file. +func readAllJSONL(t *testing.T, path string) []Event { + t.Helper() + f, err := os.Open(path) + require.NoError(t, err) + defer f.Close() + gz, err := gzip.NewReader(f) + require.NoError(t, err) + defer gz.Close() + var out []Event + sc := bufio.NewScanner(gz) + for sc.Scan() { + var ev Event + require.NoError(t, json.Unmarshal(sc.Bytes(), &ev)) + out = append(out, ev) + } + require.NoError(t, sc.Err()) + return out +} + +// listViewLogKeys walks the file store under the writer's prefix and +// returns the relative keys, sorted by filename (which is RFC3339-style +// so lex order == time order). +func listViewLogKeys(t *testing.T, root, nodeDID string) []string { + t.Helper() + dir := filepath.Join(root, "view-logs", nodeDID) + entries, err := os.ReadDir(dir) + if os.IsNotExist(err) { + return nil + } + require.NoError(t, err) + out := make([]string, 0, len(entries)) + for _, e := range entries { + if e.IsDir() || strings.HasPrefix(e.Name(), ".") { + continue + } + out = append(out, filepath.Join(dir, e.Name())) + } + sort.Strings(out) + return out +} + +func TestWriterFlushOnClose(t *testing.T) { + root := t.TempDir() + store, err := blob.NewFileStore(root) + require.NoError(t, err) + + now := time.Date(2026, 5, 17, 12, 0, 0, 0, time.UTC) + w, err := NewWriter(Config{ + Store: store, + NodeDID: "did:web:test.example", + FlushAfter: 1 * time.Hour, // won't fire during the test + Salts: NewSaltManager(newMemSaltStorage()), + Now: func() time.Time { return now }, + }) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go w.Run(ctx) + + w.Log(ctx, Event{Ts: now, Type: EventTypeManifestRequest, VideoURI: "at://x/place.stream.video/1", SID: "abc"}) + w.Log(ctx, Event{Ts: now.Add(time.Second), Type: EventTypeSegmentRequest, CID: "bafyfoo", SID: "abc"}) + + require.NoError(t, w.Close()) + + keys := listViewLogKeys(t, root, "did:web:test.example") + require.Len(t, keys, 1, "exactly one flush on Close") + got := readAllJSONL(t, keys[0]) + require.Len(t, got, 2) + require.Equal(t, EventTypeManifestRequest, got[0].Type) + require.Equal(t, "at://x/place.stream.video/1", got[0].VideoURI) + require.Equal(t, EventTypeSegmentRequest, got[1].Type) + require.Equal(t, "bafyfoo", got[1].CID) +} + +func TestWriterFlushOnSize(t *testing.T) { + root := t.TempDir() + store, err := blob.NewFileStore(root) + require.NoError(t, err) + + // Force a size flush after very few events by setting a low cap. + w, err := NewWriter(Config{ + Store: store, + NodeDID: "did:web:test.example", + FlushAfter: 1 * time.Hour, + MaxBytes: 64, // a few JSON lines compress well above this + Salts: NewSaltManager(newMemSaltStorage()), + }) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go w.Run(ctx) + + for i := 0; i < 40; i++ { + w.Log(ctx, Event{ + Ts: time.Now().UTC(), + Type: EventTypeSegmentRequest, + CID: "bafy_segment_with_a_reasonably_long_cid_string_so_we_overflow", + SID: "3jw5xxxxxxxxx", + IPHash: "deadbeefcafebabe1234567890abcdef0123456789abcdef0123456789abcdef", + OwnerDID: "did:plc:abcdefghijklmnop", + }) + } + + require.NoError(t, w.Close()) + keys := listViewLogKeys(t, root, "did:web:test.example") + require.GreaterOrEqual(t, len(keys), 2, "size cap should have forced at least one mid-stream flush before Close") + + var total int + for _, k := range keys { + total += len(readAllJSONL(t, k)) + } + require.Equal(t, 40, total, "every event lands in some flush") +} + +func TestWriterNoOpWhenEmpty(t *testing.T) { + root := t.TempDir() + store, err := blob.NewFileStore(root) + require.NoError(t, err) + + w, err := NewWriter(Config{ + Store: store, + NodeDID: "did:web:test.example", + FlushAfter: 1 * time.Hour, + Salts: NewSaltManager(newMemSaltStorage()), + }) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go w.Run(ctx) + + require.NoError(t, w.Close()) + keys := listViewLogKeys(t, root, "did:web:test.example") + require.Empty(t, keys, "no events ⇒ no files") +} + +func TestWriterConcurrentLogs(t *testing.T) { + root := t.TempDir() + store, err := blob.NewFileStore(root) + require.NoError(t, err) + + w, err := NewWriter(Config{ + Store: store, + NodeDID: "did:web:test.example", + FlushAfter: 1 * time.Hour, + Salts: NewSaltManager(newMemSaltStorage()), + }) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go w.Run(ctx) + + const goroutines = 8 + const perGoroutine = 50 + var wg sync.WaitGroup + wg.Add(goroutines) + for g := 0; g < goroutines; g++ { + go func() { + defer wg.Done() + for i := 0; i < perGoroutine; i++ { + w.Log(ctx, Event{ + Ts: time.Now().UTC(), + Type: EventTypeManifestRequest, + SID: "sid", + }) + } + }() + } + wg.Wait() + + require.NoError(t, w.Close()) + keys := listViewLogKeys(t, root, "did:web:test.example") + var total int + for _, k := range keys { + total += len(readAllJSONL(t, k)) + } + require.Equal(t, goroutines*perGoroutine, total) +}