From 9057bee003d28bc8b178c05b3009bf28f4ae73ea Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Sun, 17 May 2026 15:20:09 -0700 Subject: [PATCH] viewlog: capture manifest + segment requests for view-count aggregation Adds pkg/viewlog: async gzipped-JSONL writer that buffers playback request events and rotates them to the VOD blob.Store on a configurable interval. Each line carries the playback session id, an HMAC-anonymized IP, and per-event-type fields (video AT-URI + manifest kind for manifest fetches, blob CID + owner DID + range bytes for segment fetches). IP anonymization uses a daily-rotated salt persisted in localdb: same IP collides within a UTC day (useful for dedup), the salt changes at midnight to decorrelate across days. Pruning old salt rows is enough to render historical logs unrecoverable. HandleGetVideoPlaylist and HandleGetVideoBlob post events to the writer through a non-blocking helper; if the operator has disabled the writer (--view-log-flush-interval=0) or no VOD store is wired, the helpers are no-ops. Files land under view-logs// .jsonl.gz alongside the content blobs so they pick up the bucket's lifecycle policy for free. Setup for the aggregator + place.stream.media.viewCount publish steps in a follow-up; this commit lands only the capture half. Co-Authored-By: Claude Opus 4.7 --- pkg/api/api.go | 7 +- pkg/cmd/streamplace.go | 46 +++- pkg/config/config.go | 8 + pkg/localdb/localdb.go | 4 + pkg/localdb/view_log_salt.go | 38 ++++ pkg/spxrpc/place_stream_playback_getvideo.go | 20 ++ pkg/spxrpc/spxrpc.go | 10 +- pkg/spxrpc/viewlog_hooks.go | 64 ++++++ pkg/viewlog/event.go | 72 +++++++ pkg/viewlog/salt.go | 86 ++++++++ pkg/viewlog/salt_test.go | 122 +++++++++++ pkg/viewlog/writer.go | 214 +++++++++++++++++++ pkg/viewlog/writer_test.go | 202 +++++++++++++++++ 13 files changed, 888 insertions(+), 5 deletions(-) create mode 100644 pkg/localdb/view_log_salt.go create mode 100644 pkg/spxrpc/viewlog_hooks.go create mode 100644 pkg/viewlog/event.go create mode 100644 pkg/viewlog/salt.go create mode 100644 pkg/viewlog/salt_test.go create mode 100644 pkg/viewlog/writer.go create mode 100644 pkg/viewlog/writer_test.go 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) +} -- 2.51.2