diff --git a/api/server.go b/api/server.go index 784e25fb..60abb1d3 100644 --- a/api/server.go +++ b/api/server.go @@ -534,6 +534,7 @@ func NewApiServer(config config.Config) *ApiServer { g.Get("/users/:userId/sales/download/json", app.requireAuthForUserId, app.v1UsersSalesDownloadJson) g.Get("/users/:userId/sales/count", app.requireAuthForUserId, app.v1UsersSalesCount) g.Get("/users/:userId/sales/aggregate", app.requireAuthForUserId, app.v1UsersSalesAggregate) + g.Get("/users/:userId/listener_cities", app.requireAuthForUserId, app.v1UsersListenerCities) g.Get("/users/:userId/muted", app.v1UsersMuted) g.Get("/users/:userId/subscribers", app.v1UsersSubscribers) g.Get("/users/:userId/remixers", app.v1UsersRemixers) diff --git a/api/swagger/swagger-v1.yaml b/api/swagger/swagger-v1.yaml index fb3f1a95..c19e6497 100644 --- a/api/swagger/swagger-v1.yaml +++ b/api/swagger/swagger-v1.yaml @@ -7038,6 +7038,72 @@ paths: application/json: schema: $ref: "#/components/schemas/sales_aggregate_response" + /users/{id}/listener_cities: + get: + tags: + - users + description: + Gets the cities whose listeners played the user's tracks the most. + Plays by the user and plays with no known city are not counted. + Locations come from the listener's IP address and can be wrong for + VPN and mobile carrier traffic. Only the user or one of their + managers can call this. + operationId: Get Listener Cities + security: + - {} + - OAuth2: + - read + parameters: + - name: id + in: path + description: A User ID + required: true + schema: + type: string + - name: time + in: query + description: + Time range. week is the last 7 UTC days including today, month + the last 30. all covers every play since location rollups began + (plays older than about 400 days were already pruned). + schema: + type: string + default: month + enum: + - week + - month + - all + - name: limit + in: query + description: The number of cities to return (max 50) + schema: + type: integer + default: 10 + - name: user_id + in: query + description: The user ID of the user making the request + schema: + type: string + - name: Encoded-Data-Message + in: header + description: The data that was signed by the user for signature recovery + schema: + type: string + - name: Encoded-Data-Signature + in: header + description: The signature of data, used for signature recovery + schema: + type: string + responses: + "200": + description: Success + content: + application/json: + schema: + $ref: "#/components/schemas/listener_cities_response" + "403": + description: Not the user or one of their managers + content: {} /users/{id}/sales/download: get: tags: @@ -13761,6 +13827,13 @@ components: type: integer description: The reward amount example: 100 + listener_cities_response: + type: object + properties: + data: + type: array + items: + $ref: "#/components/schemas/listener_city" sales_aggregate_response: type: object properties: @@ -15727,6 +15800,24 @@ components: type: array items: $ref: "#/components/schemas/challenge_response" + listener_city: + required: + - city + - region + - country + - play_count + type: object + properties: + city: + type: string + region: + type: string + description: Region or state name, empty when unknown + country: + type: string + description: Country name, empty when unknown + play_count: + type: integer sales_aggregate: required: - content_id diff --git a/api/v1_users_listener_cities.go b/api/v1_users_listener_cities.go new file mode 100644 index 00000000..432a6782 --- /dev/null +++ b/api/v1_users_listener_cities.go @@ -0,0 +1,69 @@ +package api + +import ( + "time" + + "github.com/gofiber/fiber/v2" + "github.com/jackc/pgx/v5" +) + +type GetUsersListenerCitiesQueryParams struct { + Time string `query:"time" default:"month" validate:"oneof=week month all"` + Limit int `query:"limit" default:"10" validate:"min=1,max=50"` +} + +type ListenerCity struct { + City string `json:"city" db:"city"` + Region string `json:"region" db:"region"` + Country string `json:"country" db:"country"` + PlayCount int64 `json:"play_count" db:"play_count"` +} + +// v1UsersListenerCities returns the cities that played the artist's tracks +// the most, from aggregate_daily_city_plays (see jobs.CityPlaysJob). The +// artist's own plays and plays with no city are not counted. +func (app *ApiServer) v1UsersListenerCities(c *fiber.Ctx) error { + params := GetUsersListenerCitiesQueryParams{} + if err := app.ParseAndValidateQueryParams(c, ¶ms); err != nil { + return err + } + + // Days are UTC; "week" is today plus the 6 days before it. + var since *time.Time + today := time.Now().UTC().Truncate(24 * time.Hour) + switch params.Time { + case "week": + t := today.AddDate(0, 0, -6) + since = &t + case "month": + t := today.AddDate(0, 0, -29) + since = &t + } + + rows, err := app.pool.Query(c.Context(), ` + SELECT a.city, a.region, a.country, SUM(a.count)::bigint AS play_count + FROM tracks t + JOIN aggregate_daily_city_plays a ON a.play_item_id = t.track_id + WHERE t.owner_id = @userId + AND t.is_current + AND (@since::date IS NULL OR a.day >= @since::date) + GROUP BY a.city, a.region, a.country + ORDER BY play_count DESC, a.city, a.region, a.country + LIMIT @limit + `, pgx.NamedArgs{ + "userId": app.getUserId(c), + "since": since, + "limit": params.Limit, + }) + if err != nil { + return err + } + cities, err := pgx.CollectRows(rows, pgx.RowToStructByName[ListenerCity]) + if err != nil { + return err + } + + return c.JSON(fiber.Map{ + "data": cities, + }) +} diff --git a/api/v1_users_listener_cities_test.go b/api/v1_users_listener_cities_test.go new file mode 100644 index 00000000..b1f9252f --- /dev/null +++ b/api/v1_users_listener_cities_test.go @@ -0,0 +1,107 @@ +package api + +import ( + "testing" + "time" + + "api.audius.co/database" + "api.audius.co/trashid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestV1UsersListenerCities(t *testing.T) { + artistWallet := "0x7d273271690538cf855e5b3002a0dd8c154bb060" + managerWallet := "0x4954d18926ba0ed9378938444731be4e622537b2" + otherWallet := "0xc3d1d41e6872ffbd15c473d14fc3a9250be5b5e0" + app := emptyTestApp(t) + + today := time.Now().UTC().Truncate(24 * time.Hour) + twoWeeksAgo := today.AddDate(0, 0, -14) + twoMonthsAgo := today.AddDate(0, -2, 0) + + database.Seed(app.pool.Replicas[0], database.FixtureMap{ + "users": { + {"user_id": 1, "wallet": artistWallet, "handle": "artist", "handle_lc": "artist"}, + {"user_id": 2, "wallet": otherWallet, "handle": "other", "handle_lc": "other"}, + {"user_id": 3, "wallet": managerWallet, "handle": "manager", "handle_lc": "manager"}, + }, + "grants": { + { + "user_id": 1, + "grantee_address": managerWallet, + "is_current": true, + "is_approved": true, + "is_revoked": false, + "created_at": time.Now(), + "updated_at": time.Now(), + }, + }, + "tracks": { + {"track_id": 100, "owner_id": 1, "title": "A"}, + {"track_id": 101, "owner_id": 1, "title": "B"}, + {"track_id": 200, "owner_id": 2, "title": "Someone else's"}, + }, + "aggregate_daily_city_plays": { + // Atlanta: 3 this week across two tracks, 5 two months ago. + {"play_item_id": 100, "day": today, "city": "Atlanta", "region": "Georgia", "country": "United States", "count": 2}, + {"play_item_id": 101, "day": today, "city": "Atlanta", "region": "Georgia", "country": "United States", "count": 1}, + {"play_item_id": 100, "day": twoMonthsAgo, "city": "Atlanta", "region": "Georgia", "country": "United States", "count": 5}, + // Berlin: 2 this week. + {"play_item_id": 100, "day": today, "city": "Berlin", "region": "Land Berlin", "country": "Germany", "count": 2}, + // Lagos: 4 two weeks ago. + {"play_item_id": 101, "day": twoWeeksAgo, "city": "Lagos", "region": "Lagos", "country": "Nigeria", "count": 4}, + // Another artist's track is not counted. + {"play_item_id": 200, "day": today, "city": "Tokyo", "region": "Tokyo", "country": "Japan", "count": 50}, + }, + }) + + var res struct { + Data []ListenerCity + } + path := "/v1/users/" + trashid.MustEncodeHashID(1) + "/listener_cities" + + t.Run("week", func(t *testing.T) { + status, _ := testGetWithWallet(t, app, path+"?time=week", artistWallet, &res) + require.Equal(t, 200, status) + assert.Equal(t, []ListenerCity{ + {City: "Atlanta", Region: "Georgia", Country: "United States", PlayCount: 3}, + {City: "Berlin", Region: "Land Berlin", Country: "Germany", PlayCount: 2}, + }, res.Data) + }) + + t.Run("month is the default", func(t *testing.T) { + status, _ := testGetWithWallet(t, app, path, artistWallet, &res) + require.Equal(t, 200, status) + assert.Equal(t, []ListenerCity{ + {City: "Lagos", Region: "Lagos", Country: "Nigeria", PlayCount: 4}, + {City: "Atlanta", Region: "Georgia", Country: "United States", PlayCount: 3}, + {City: "Berlin", Region: "Land Berlin", Country: "Germany", PlayCount: 2}, + }, res.Data) + }) + + t.Run("all with limit", func(t *testing.T) { + status, _ := testGetWithWallet(t, app, path+"?time=all&limit=2", artistWallet, &res) + require.Equal(t, 200, status) + assert.Equal(t, []ListenerCity{ + {City: "Atlanta", Region: "Georgia", Country: "United States", PlayCount: 8}, + {City: "Lagos", Region: "Lagos", Country: "Nigeria", PlayCount: 4}, + }, res.Data) + }) + + t.Run("manager can read", func(t *testing.T) { + status, _ := testGetWithWallet(t, app, path+"?time=week", managerWallet, &res) + require.Equal(t, 200, status) + assert.Len(t, res.Data, 2) + }) + + t.Run("other users cannot", func(t *testing.T) { + status, _ := testGetWithWallet(t, app, path, otherWallet) + assert.Equal(t, 403, status) + }) + + t.Run("bad time", func(t *testing.T) { + status, _ := testGetWithWallet(t, app, path+"?time=year", artistWallet) + assert.Equal(t, 400, status) + }) +} diff --git a/database/seed.go b/database/seed.go index 1350c0f8..af350258 100644 --- a/database/seed.go +++ b/database/seed.go @@ -397,6 +397,14 @@ var ( "count": 0, "country": "US", }, + "aggregate_daily_city_plays": { + "play_item_id": nil, + "day": time.Now(), + "country": "", + "region": "", + "city": "", + "count": 1, + }, "core_indexed_blocks": { "height": 1, "chain_id": "audius-mainnet-alpha-beta", diff --git a/ddl/migrations/0245_aggregate_daily_city_plays.sql b/ddl/migrations/0245_aggregate_daily_city_plays.sql new file mode 100644 index 00000000..5927b871 --- /dev/null +++ b/ddl/migrations/0245_aggregate_daily_city_plays.sql @@ -0,0 +1,24 @@ +-- Plays per track per day per listener city, for the Artist Dashboard's +-- top cities (GET /v1/users/:id/listener_cities). +-- +-- plays has no index by track, so per-artist location queries can't read it +-- directly. jobs.CityPlaysJob fills this table from plays (checkpointed by +-- plays.id), skipping plays with no city and plays by the track's owner. +-- The first runs backfill whatever plays are still retained (~400 days). + +BEGIN; + +CREATE TABLE IF NOT EXISTS aggregate_daily_city_plays ( + play_item_id INTEGER NOT NULL, + day DATE NOT NULL, + country TEXT NOT NULL, + region TEXT NOT NULL, + city TEXT NOT NULL, + count INTEGER NOT NULL, + PRIMARY KEY (play_item_id, day, country, region, city) +); + +COMMENT ON TABLE aggregate_daily_city_plays IS + 'Plays per track per UTC day per listener city, excluding the track owner''s own plays and plays with no city. Filled by jobs.CityPlaysJob.'; + +COMMIT; diff --git a/indexer/indexer.go b/indexer/indexer.go index 3db5d10e..e97a9fad 100644 --- a/indexer/indexer.go +++ b/indexer/indexer.go @@ -206,6 +206,10 @@ func (ci *CoreIndexer) startParityJobs(ctx context.Context) { jobs.NewPrunePlaysJob(ci.Config, ci.pool). ScheduleEvery(ctx, 30*time.Second) + // Backfills 250k play ids per run, then keeps up with new plays. + jobs.NewCityPlaysJob(ci.Config, ci.pool). + ScheduleEvery(ctx, 15*time.Second) + jobs.NewUserListeningHistoryJob(ci.Config, ci.pool). ScheduleEvery(ctx, 5*time.Second) diff --git a/jobs/index_city_plays.go b/jobs/index_city_plays.go new file mode 100644 index 00000000..b7684264 --- /dev/null +++ b/jobs/index_city_plays.go @@ -0,0 +1,154 @@ +package jobs + +import ( + "context" + "fmt" + "sync" + "time" + + "api.audius.co/config" + "api.audius.co/database" + "api.audius.co/logging" + "github.com/jackc/pgx/v5" + "go.uber.org/zap" +) + +// CityPlaysJob rolls plays up into aggregate_daily_city_plays, which serves +// GET /v1/users/:id/listener_cities. +// +// Each run reads plays after the last checkpointed plays.id, at most +// batchSize ids at a time so the first runs can backfill the retained plays +// without one huge statement. Plays with no city and plays by the track's +// owner are skipped, matching the weekly artist recap. +type CityPlaysJob struct { + pool database.DbPool + logger *zap.Logger + batchSize int64 + + mutex sync.Mutex + isRunning bool +} + +const ( + CityPlaysCheckpoint = "aggregate_daily_city_plays" + CityPlaysBatchSize = 250_000 +) + +func NewCityPlaysJob(cfg config.Config, pool database.DbPool) *CityPlaysJob { + return &CityPlaysJob{ + pool: pool, + logger: logging.NewZapLogger(cfg).Named("CityPlaysJob"), + batchSize: CityPlaysBatchSize, + } +} + +// WithBatchSize overrides the per-run id range (for tests). +func (j *CityPlaysJob) WithBatchSize(n int64) *CityPlaysJob { + j.batchSize = n + return j +} + +func (j *CityPlaysJob) ScheduleEvery(ctx context.Context, interval time.Duration) *CityPlaysJob { + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ticker.C: + j.Run(ctx) + case <-ctx.Done(): + j.logger.Info("Job shutting down") + return + } + } + }() + return j +} + +func (j *CityPlaysJob) Run(ctx context.Context) { + if err := j.run(ctx); err != nil { + j.logger.Error("Job run failed", zap.Error(err)) + } +} + +func (j *CityPlaysJob) run(ctx context.Context) error { + start := time.Now() + j.mutex.Lock() + if j.isRunning { + j.mutex.Unlock() + return fmt.Errorf("job is already running") + } + j.isRunning = true + j.mutex.Unlock() + defer func() { + j.mutex.Lock() + j.isRunning = false + j.mutex.Unlock() + }() + + prev, err := getCheckpoint(ctx, j.pool, CityPlaysCheckpoint) + if err != nil { + return fmt.Errorf("read checkpoint: %w", err) + } + + var minId, maxId *int64 + if err := j.pool.QueryRow(ctx, "SELECT MIN(id), MAX(id) FROM plays").Scan(&minId, &maxId); err != nil { + return fmt.Errorf("read plays id range: %w", err) + } + if maxId == nil || *maxId <= prev { + return nil + } + // Skip ids that have already been pruned. + if *minId-1 > prev { + prev = *minId - 1 + } + end := min(prev+j.batchSize, *maxId) + + // The counts and the checkpoint commit together, so a failed run is + // retried from the same place instead of counting plays twice. + tx, err := j.pool.Begin(ctx) + if err != nil { + return err + } + defer tx.Rollback(ctx) + + res, err := tx.Exec(ctx, ` + INSERT INTO aggregate_daily_city_plays (play_item_id, day, country, region, city, count) + SELECT + p.play_item_id, + p.created_at::date, + COALESCE(TRIM(p.country), ''), + COALESCE(TRIM(p.region), ''), + TRIM(p.city), + COUNT(*) + FROM plays p + JOIN tracks t ON t.track_id = p.play_item_id AND t.is_current + WHERE p.id > @prev AND p.id <= @end + AND NULLIF(TRIM(p.city), '') IS NOT NULL + AND p.user_id IS DISTINCT FROM t.owner_id + GROUP BY 1, 2, 3, 4, 5 + ON CONFLICT (play_item_id, day, country, region, city) + DO UPDATE SET count = aggregate_daily_city_plays.count + EXCLUDED.count + `, pgx.NamedArgs{"prev": prev, "end": end}) + if err != nil { + return fmt.Errorf("upsert city plays: %w", err) + } + + if _, err := tx.Exec(ctx, ` + INSERT INTO indexing_checkpoints (tablename, last_checkpoint) + VALUES ($1, $2) + ON CONFLICT (tablename) DO UPDATE SET last_checkpoint = EXCLUDED.last_checkpoint + `, CityPlaysCheckpoint, end); err != nil { + return fmt.Errorf("save checkpoint: %w", err) + } + if err := tx.Commit(ctx); err != nil { + return err + } + + j.logger.Debug("City plays updated", + zap.Int64("prev_checkpoint", prev), + zap.Int64("new_checkpoint", end), + zap.Int64("rows", res.RowsAffected()), + zap.Duration("duration", time.Since(start))) + return nil +} diff --git a/jobs/index_city_plays_test.go b/jobs/index_city_plays_test.go new file mode 100644 index 00000000..51c40d51 --- /dev/null +++ b/jobs/index_city_plays_test.go @@ -0,0 +1,95 @@ +package jobs + +import ( + "context" + "testing" + "time" + + "api.audius.co/database" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestCityPlaysJob(t *testing.T) { + pool := database.CreateTestDatabase(t, "test_jobs") + defer pool.Close() + ctx := context.Background() + + day1 := time.Date(2026, 11, 2, 10, 0, 0, 0, time.UTC) + day2 := time.Date(2026, 11, 3, 23, 30, 0, 0, time.UTC) + + database.Seed(pool, database.FixtureMap{ + "users": { + {"user_id": 1, "wallet": "0x01"}, + {"user_id": 2, "wallet": "0x02"}, + }, + "tracks": { + {"track_id": 100, "owner_id": 1, "title": "T"}, + }, + "plays": { + {"id": 1, "user_id": 2, "play_item_id": 100, "created_at": day1, "city": "Atlanta", "region": "Georgia", "country": "United States"}, + {"id": 2, "user_id": nil, "play_item_id": 100, "created_at": day1, "city": " Atlanta ", "region": "Georgia", "country": "United States"}, + // The owner's own play is skipped. + {"id": 3, "user_id": 1, "play_item_id": 100, "created_at": day1, "city": "Atlanta", "region": "Georgia", "country": "United States"}, + // No city: skipped. + {"id": 4, "user_id": 2, "play_item_id": 100, "created_at": day1, "city": "", "country": "United States"}, + {"id": 5, "user_id": 2, "play_item_id": 100, "created_at": day1, "city": nil}, + {"id": 6, "user_id": 2, "play_item_id": 100, "created_at": day2, "city": "Atlanta", "region": "Georgia", "country": "United States"}, + {"id": 7, "user_id": 2, "play_item_id": 100, "created_at": day2, "city": "Berlin", "region": "Land Berlin", "country": "Germany"}, + }, + }) + + // Batches of 3 ids: three runs to catch up, a fourth is a no-op. + job := NewCityPlaysJob(newTestConfig(), pool).WithBatchSize(3) + for range 4 { + require.NoError(t, job.run(ctx)) + } + + type row struct { + Day string + City string + Count int32 + } + var rows []row + r, err := pool.Query(ctx, ` + SELECT day::text, city, count FROM aggregate_daily_city_plays + WHERE play_item_id = 100 ORDER BY day, city`) + require.NoError(t, err) + for r.Next() { + var x row + require.NoError(t, r.Scan(&x.Day, &x.City, &x.Count)) + rows = append(rows, x) + } + r.Close() + + assert.Equal(t, []row{ + {"2026-11-02", "Atlanta", 2}, + {"2026-11-03", "Atlanta", 1}, + {"2026-11-03", "Berlin", 1}, + }, rows) + + cp, err := getCheckpoint(ctx, pool, CityPlaysCheckpoint) + require.NoError(t, err) + assert.Equal(t, int64(7), cp) +} + +func TestCityPlaysJob_SkipsPrunedIds(t *testing.T) { + pool := database.CreateTestDatabase(t, "test_jobs") + defer pool.Close() + ctx := context.Background() + + database.Seed(pool, database.FixtureMap{ + "users": {{"user_id": 1, "wallet": "0x01"}}, + "tracks": {{"track_id": 100, "owner_id": 1, "title": "T"}}, + "plays": { + {"id": 1_000_000, "play_item_id": 100, "created_at": time.Now(), "city": "Lagos", "region": "Lagos", "country": "Nigeria"}, + }, + }) + + job := NewCityPlaysJob(newTestConfig(), pool).WithBatchSize(10) + require.NoError(t, job.run(ctx)) + + var count int32 + require.NoError(t, pool.QueryRow(ctx, "SELECT SUM(count) FROM aggregate_daily_city_plays").Scan(&count)) + assert.Equal(t, int32(1), count) +} diff --git a/sql/01_schema.sql b/sql/01_schema.sql index d36dbc8d..321afe84 100644 --- a/sql/01_schema.sql +++ b/sql/01_schema.sql @@ -7061,6 +7061,27 @@ CREATE SEQUENCE public.aggregate_monthly_app_name_metrics_id_seq ALTER SEQUENCE public.aggregate_monthly_app_name_metrics_id_seq OWNED BY public.aggregate_monthly_app_name_metrics.id; +-- +-- Name: aggregate_daily_city_plays; Type: TABLE; Schema: public; Owner: - +-- + +CREATE TABLE public.aggregate_daily_city_plays ( + play_item_id integer NOT NULL, + day date NOT NULL, + country text NOT NULL, + region text NOT NULL, + city text NOT NULL, + count integer NOT NULL +); + + +-- +-- Name: TABLE aggregate_daily_city_plays; Type: COMMENT; Schema: public; Owner: - +-- + +COMMENT ON TABLE public.aggregate_daily_city_plays IS 'Plays per track per UTC day per listener city, excluding the track owner''s own plays and plays with no city. Filled by jobs.CityPlaysJob.'; + + -- -- Name: aggregate_monthly_plays; Type: TABLE; Schema: public; Owner: - -- @@ -11920,6 +11941,14 @@ ALTER TABLE ONLY public.aggregate_monthly_app_name_metrics ADD CONSTRAINT aggregate_monthly_app_name_metrics_pkey PRIMARY KEY (id); +-- +-- Name: aggregate_daily_city_plays aggregate_daily_city_plays_pkey; Type: CONSTRAINT; Schema: public; Owner: - +-- + +ALTER TABLE ONLY public.aggregate_daily_city_plays + ADD CONSTRAINT aggregate_daily_city_plays_pkey PRIMARY KEY (play_item_id, day, country, region, city); + + -- -- Name: aggregate_monthly_plays aggregate_monthly_plays_pkey; Type: CONSTRAINT; Schema: public; Owner: - --