feat(api): top listener cities for the artist dashboard

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)

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

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, &params); 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,
+	})
+}

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)
+	})
+}

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",

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;

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)