From 52975573c4ab7c2f85d1cf3fe700be6e0d567b53 Mon Sep 17 00:00:00 2001 From: Dylan Jeffers Date: Wed, 11 Nov 2026 16:01:09 -0800 Subject: [PATCH 1/2] feat(api): top listener cities for the artist dashboard GET /v1/users/:id/listener_cities?time=week|month|all returns the cities whose listeners played the artist's tracks the most. Auth: the artist or a manager. plays has no index by track, so a new indexer job rolls plays up into aggregate_daily_city_plays (per track, UTC day and city), skipping the owner's own plays and plays with no city. The job is checkpointed on plays.id and backfills retained plays 250k ids per run. Co-Authored-By: Claude Opus 5.5 --- api/server.go | 1 + api/swagger/swagger-v1.yaml | 91 ++++++++++++ api/v1_users_listener_cities.go | 69 +++++++++ api/v1_users_listener_cities_test.go | 107 ++++++++++++++ database/seed.go | 8 + .../0245_aggregate_daily_city_plays.sql | 24 +++ indexer/indexer.go | 4 + jobs/index_city_plays.go | 139 ++++++++++++++++++ jobs/index_city_plays_test.go | 95 ++++++++++++ sql/01_schema.sql | 29 ++++ 10 files changed, 567 insertions(+) create mode 100644 api/v1_users_listener_cities.go create mode 100644 api/v1_users_listener_cities_test.go create mode 100644 ddl/migrations/0245_aggregate_daily_city_plays.sql create mode 100644 jobs/index_city_plays.go create mode 100644 jobs/index_city_plays_test.go 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..9a5bbfef --- /dev/null +++ b/jobs/index_city_plays.go @@ -0,0 +1,139 @@ +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) + + res, err := j.pool.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 := saveCheckpoint(ctx, j.pool, CityPlaysCheckpoint, end); err != nil { + return fmt.Errorf("save checkpoint: %w", 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: - -- From 9c9f7a500c5cdfa6d26f9fc1bcc0d370cba185ed Mon Sep 17 00:00:00 2001 From: Dylan Jeffers Date: Wed, 11 Nov 2026 16:35:34 -0800 Subject: [PATCH 2/2] fix(api): commit city play counts and checkpoint together Co-Authored-By: Claude Opus 5.5 --- jobs/index_city_plays.go | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/jobs/index_city_plays.go b/jobs/index_city_plays.go index 9a5bbfef..b7684264 100644 --- a/jobs/index_city_plays.go +++ b/jobs/index_city_plays.go @@ -104,7 +104,15 @@ func (j *CityPlaysJob) run(ctx context.Context) error { } end := min(prev+j.batchSize, *maxId) - res, err := j.pool.Exec(ctx, ` + // 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, @@ -126,9 +134,16 @@ func (j *CityPlaysJob) run(ctx context.Context) error { return fmt.Errorf("upsert city plays: %w", err) } - if err := saveCheckpoint(ctx, j.pool, CityPlaysCheckpoint, end); err != nil { + 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),