From ce5ce141f8595f51296f0a3055f322310a13e60e Mon Sep 17 00:00:00 2001 From: DiheChen Date: Wed, 11 Nov 2026 00:05:03 +0800 Subject: [PATCH] feat: keep each member's listening history and move a guest's history into the account on login --- db/migrations/00005_listens.sql | 21 ++ db/queries/listens.sql | 57 +++ db/queries/members.sql | 4 + .../postgres/internal/query/listens.sql.go | 234 ++++++++++++ .../postgres/internal/query/members.sql.go | 24 +- .../out/postgres/internal/query/models.go | 15 +- internal/adapter/out/postgres/listens.go | 119 ++++++ internal/adapter/out/postgres/listens_test.go | 346 ++++++++++++++++++ internal/adapter/out/postgres/members.go | 14 +- internal/app/identity/account_test.go | 25 ++ internal/app/port/listen.go | 36 ++ internal/app/port/member.go | 2 + internal/app/port/porttest/listen.go | 91 +++++ internal/app/port/porttest/store.go | 17 +- internal/app/port/porttest/store_test.go | 24 ++ 15 files changed, 1013 insertions(+), 16 deletions(-) create mode 100644 db/migrations/00005_listens.sql create mode 100644 db/queries/listens.sql create mode 100644 internal/adapter/out/postgres/internal/query/listens.sql.go create mode 100644 internal/adapter/out/postgres/listens.go create mode 100644 internal/adapter/out/postgres/listens_test.go create mode 100644 internal/app/port/listen.go create mode 100644 internal/app/port/porttest/listen.go diff --git a/db/migrations/00005_listens.sql b/db/migrations/00005_listens.sql new file mode 100644 index 0000000..f3eb6c0 --- /dev/null +++ b/db/migrations/00005_listens.sql @@ -0,0 +1,21 @@ +-- +goose Up +-- 关闭之后不再记录新的播放历史，已有的保留。 +ALTER TABLE members ADD COLUMN record_listens boolean NOT NULL DEFAULT true; + +-- 成员在房间里听到过的歌，每首只留最近一次。元数据存一份快照，不关联 tracks：管理员删除已存储的媒体时会连带删掉 tracks 中对应的行。 +CREATE TABLE listens ( + member_id uuid NOT NULL REFERENCES members ON DELETE CASCADE, + source text NOT NULL, + track_id text NOT NULL, + title text NOT NULL, + artist text, + album text, + cover_url text, + cover_key text, + duration_ms bigint NOT NULL, + page_url text NOT NULL, + heard_at timestamptz NOT NULL, + PRIMARY KEY (member_id, source, track_id) +); + +CREATE INDEX ON listens (member_id, heard_at DESC); diff --git a/db/queries/listens.sql b/db/queries/listens.sql new file mode 100644 index 0000000..789f779 --- /dev/null +++ b/db/queries/listens.sql @@ -0,0 +1,57 @@ +-- name: LockListeners :many +-- 按 ID 的顺序锁住开启了记录的成员：同一个成员的历史依次写入，删掉超出上限的部分时不会多删或少删； +-- 加锁的顺序固定，同时写几个成员的事务之间不会死锁。 +SELECT id FROM members WHERE id = ANY(@ids::uuid[]) AND record_listens ORDER BY id FOR NO KEY UPDATE; + +-- name: UpsertListens :exec +INSERT INTO listens (member_id, source, track_id, title, artist, album, cover_url, cover_key, duration_ms, page_url, heard_at) +SELECT m, @source::text, @track_id::text, @title::text, sqlc.narg(artist)::text, sqlc.narg(album)::text, sqlc.narg(cover_url)::text, + sqlc.narg(cover_key)::text, @duration_ms::bigint, @page_url::text, @heard_at::timestamptz +FROM unnest(@member_ids::uuid[]) AS m +ON CONFLICT (member_id, source, track_id) DO UPDATE +SET title = excluded.title, artist = excluded.artist, album = excluded.album, cover_url = excluded.cover_url, + cover_key = excluded.cover_key, duration_ms = excluded.duration_ms, page_url = excluded.page_url, + heard_at = greatest(listens.heard_at, excluded.heard_at); + +-- name: MoveListens :exec +-- 并入的规则与 UpsertListens 相同，并入的那份算新写入的；to_id 关闭了记录时只删不并。 +WITH moved AS ( + DELETE FROM listens WHERE member_id = sqlc.arg(from_id)::uuid + RETURNING source, track_id, title, artist, album, cover_url, cover_key, duration_ms, page_url, heard_at +) +INSERT INTO listens (member_id, source, track_id, title, artist, album, cover_url, cover_key, duration_ms, page_url, heard_at) +SELECT sqlc.arg(to_id)::uuid, source, track_id, title, artist, album, cover_url, cover_key, duration_ms, page_url, heard_at +FROM moved +WHERE EXISTS (SELECT FROM members WHERE id = sqlc.arg(to_id)::uuid AND record_listens) +ON CONFLICT (member_id, source, track_id) DO UPDATE +SET title = excluded.title, artist = excluded.artist, album = excluded.album, cover_url = excluded.cover_url, + cover_key = excluded.cover_key, duration_ms = excluded.duration_ms, page_url = excluded.page_url, + heard_at = greatest(listens.heard_at, excluded.heard_at); + +-- name: TrimListens :exec +-- 排序与 MemberListens 相同：留下的正是读取时返回的那些。 +DELETE FROM listens l +USING ( + SELECT member_id, source, track_id, + row_number() OVER (PARTITION BY member_id ORDER BY heard_at DESC, source, track_id) AS n + FROM listens WHERE member_id = ANY(@member_ids::uuid[]) +) ranked +WHERE l.member_id = ranked.member_id AND l.source = ranked.source AND l.track_id = ranked.track_id AND ranked.n > @keep::bigint; + +-- name: MemberListens :many +SELECT source, track_id, title, artist, album, cover_url, cover_key, duration_ms, page_url, heard_at +FROM listens WHERE member_id = @member_id +ORDER BY heard_at DESC, source, track_id +LIMIT @max; + +-- name: DeleteListen :exec +DELETE FROM listens WHERE member_id = @member_id AND source = @source AND track_id = @track_id; + +-- name: ClearListens :exec +DELETE FROM listens WHERE member_id = @member_id; + +-- name: ListenRecording :one +SELECT record_listens FROM members WHERE id = @id; + +-- name: SetListenRecording :execrows +UPDATE members SET record_listens = @record_listens WHERE id = @id; diff --git a/db/queries/members.sql b/db/queries/members.sql index 010f8b5..21f2820 100644 --- a/db/queries/members.sql +++ b/db/queries/members.sql @@ -40,11 +40,15 @@ SELECT EXISTS (SELECT FROM member_sessions WHERE token_hash = @token_hash AND me -- name: LockMemberPassword :one -- 让修改密码等到这次登录提交之后，它随后删除会话时会删掉这次登录签发的会话。 +-- 用互斥的 FOR NO KEY UPDATE，不用 FOR SHARE：同一个账号的几次登录依次并入游客的历史，否则两边会按相反的顺序抢同一批历史行而死锁。 SELECT coalesce(password_hash, '')::text AS password_hash FROM members WHERE id = @id FOR NO KEY UPDATE; -- name: DeleteSession :execrows DELETE FROM member_sessions WHERE token_hash = @token_hash; +-- name: TakeSession :one +DELETE FROM member_sessions WHERE token_hash = @token_hash RETURNING member_id; + -- name: DeleteMemberSessions :many -- keep 为 NULL 时删除这个成员的全部会话。 DELETE FROM member_sessions WHERE member_id = @member_id AND token_hash IS DISTINCT FROM sqlc.narg(keep)::bytea diff --git a/internal/adapter/out/postgres/internal/query/listens.sql.go b/internal/adapter/out/postgres/internal/query/listens.sql.go new file mode 100644 index 0000000..cdd460c --- /dev/null +++ b/internal/adapter/out/postgres/internal/query/listens.sql.go @@ -0,0 +1,234 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: listens.sql + +package query + +import ( + "context" + "time" +) + +const clearListens = `-- name: ClearListens :exec +DELETE FROM listens WHERE member_id = $1 +` + +func (q *Queries) ClearListens(ctx context.Context, memberID string) error { + _, err := q.db.Exec(ctx, clearListens, memberID) + return err +} + +const deleteListen = `-- name: DeleteListen :exec +DELETE FROM listens WHERE member_id = $1 AND source = $2 AND track_id = $3 +` + +type DeleteListenParams struct { + MemberID string + Source string + TrackID string +} + +func (q *Queries) DeleteListen(ctx context.Context, arg DeleteListenParams) error { + _, err := q.db.Exec(ctx, deleteListen, arg.MemberID, arg.Source, arg.TrackID) + return err +} + +const listenRecording = `-- name: ListenRecording :one +SELECT record_listens FROM members WHERE id = $1 +` + +func (q *Queries) ListenRecording(ctx context.Context, id string) (bool, error) { + row := q.db.QueryRow(ctx, listenRecording, id) + var record_listens bool + err := row.Scan(&record_listens) + return record_listens, err +} + +const lockListeners = `-- name: LockListeners :many +SELECT id FROM members WHERE id = ANY($1::uuid[]) AND record_listens ORDER BY id FOR NO KEY UPDATE +` + +// 按 ID 的顺序锁住开启了记录的成员：同一个成员的历史依次写入，删掉超出上限的部分时不会多删或少删； +// 加锁的顺序固定，同时写几个成员的事务之间不会死锁。 +func (q *Queries) LockListeners(ctx context.Context, ids []string) ([]string, error) { + rows, err := q.db.Query(ctx, lockListeners, ids) + if err != nil { + return nil, err + } + defer rows.Close() + var items []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, err + } + items = append(items, id) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const memberListens = `-- name: MemberListens :many +SELECT source, track_id, title, artist, album, cover_url, cover_key, duration_ms, page_url, heard_at +FROM listens WHERE member_id = $1 +ORDER BY heard_at DESC, source, track_id +LIMIT $2 +` + +type MemberListensParams struct { + MemberID string + Max int32 +} + +type MemberListensRow struct { + Source string + TrackID string + Title string + Artist *string + Album *string + CoverURL *string + CoverKey *string + DurationMs int64 + PageURL string + HeardAt time.Time +} + +func (q *Queries) MemberListens(ctx context.Context, arg MemberListensParams) ([]MemberListensRow, error) { + rows, err := q.db.Query(ctx, memberListens, arg.MemberID, arg.Max) + if err != nil { + return nil, err + } + defer rows.Close() + var items []MemberListensRow + for rows.Next() { + var i MemberListensRow + if err := rows.Scan( + &i.Source, + &i.TrackID, + &i.Title, + &i.Artist, + &i.Album, + &i.CoverURL, + &i.CoverKey, + &i.DurationMs, + &i.PageURL, + &i.HeardAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const moveListens = `-- name: MoveListens :exec +WITH moved AS ( + DELETE FROM listens WHERE member_id = $2::uuid + RETURNING source, track_id, title, artist, album, cover_url, cover_key, duration_ms, page_url, heard_at +) +INSERT INTO listens (member_id, source, track_id, title, artist, album, cover_url, cover_key, duration_ms, page_url, heard_at) +SELECT $1::uuid, source, track_id, title, artist, album, cover_url, cover_key, duration_ms, page_url, heard_at +FROM moved +WHERE EXISTS (SELECT FROM members WHERE id = $1::uuid AND record_listens) +ON CONFLICT (member_id, source, track_id) DO UPDATE +SET title = excluded.title, artist = excluded.artist, album = excluded.album, cover_url = excluded.cover_url, + cover_key = excluded.cover_key, duration_ms = excluded.duration_ms, page_url = excluded.page_url, + heard_at = greatest(listens.heard_at, excluded.heard_at) +` + +type MoveListensParams struct { + ToID string + FromID string +} + +// 并入的规则与 UpsertListens 相同，并入的那份算新写入的；to_id 关闭了记录时只删不并。 +func (q *Queries) MoveListens(ctx context.Context, arg MoveListensParams) error { + _, err := q.db.Exec(ctx, moveListens, arg.ToID, arg.FromID) + return err +} + +const setListenRecording = `-- name: SetListenRecording :execrows +UPDATE members SET record_listens = $1 WHERE id = $2 +` + +type SetListenRecordingParams struct { + RecordListens bool + ID string +} + +func (q *Queries) SetListenRecording(ctx context.Context, arg SetListenRecordingParams) (int64, error) { + result, err := q.db.Exec(ctx, setListenRecording, arg.RecordListens, arg.ID) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + +const trimListens = `-- name: TrimListens :exec +DELETE FROM listens l +USING ( + SELECT member_id, source, track_id, + row_number() OVER (PARTITION BY member_id ORDER BY heard_at DESC, source, track_id) AS n + FROM listens WHERE member_id = ANY($1::uuid[]) +) ranked +WHERE l.member_id = ranked.member_id AND l.source = ranked.source AND l.track_id = ranked.track_id AND ranked.n > $2::bigint +` + +type TrimListensParams struct { + MemberIds []string + Keep int64 +} + +// 排序与 MemberListens 相同：留下的正是读取时返回的那些。 +func (q *Queries) TrimListens(ctx context.Context, arg TrimListensParams) error { + _, err := q.db.Exec(ctx, trimListens, arg.MemberIds, arg.Keep) + return err +} + +const upsertListens = `-- name: UpsertListens :exec +INSERT INTO listens (member_id, source, track_id, title, artist, album, cover_url, cover_key, duration_ms, page_url, heard_at) +SELECT m, $1::text, $2::text, $3::text, $4::text, $5::text, $6::text, + $7::text, $8::bigint, $9::text, $10::timestamptz +FROM unnest($11::uuid[]) AS m +ON CONFLICT (member_id, source, track_id) DO UPDATE +SET title = excluded.title, artist = excluded.artist, album = excluded.album, cover_url = excluded.cover_url, + cover_key = excluded.cover_key, duration_ms = excluded.duration_ms, page_url = excluded.page_url, + heard_at = greatest(listens.heard_at, excluded.heard_at) +` + +type UpsertListensParams struct { + Source string + TrackID string + Title string + Artist *string + Album *string + CoverURL *string + CoverKey *string + DurationMs int64 + PageURL string + HeardAt time.Time + MemberIds []string +} + +func (q *Queries) UpsertListens(ctx context.Context, arg UpsertListensParams) error { + _, err := q.db.Exec(ctx, upsertListens, + arg.Source, + arg.TrackID, + arg.Title, + arg.Artist, + arg.Album, + arg.CoverURL, + arg.CoverKey, + arg.DurationMs, + arg.PageURL, + arg.HeardAt, + arg.MemberIds, + ) + return err +} diff --git a/internal/adapter/out/postgres/internal/query/members.sql.go b/internal/adapter/out/postgres/internal/query/members.sql.go index e864f91..db1512a 100644 --- a/internal/adapter/out/postgres/internal/query/members.sql.go +++ b/internal/adapter/out/postgres/internal/query/members.sql.go @@ -10,7 +10,7 @@ import ( ) const accountByUsername = `-- name: AccountByUsername :one -SELECT id, name, avatar_hash, created_at, username, password_hash, email FROM members WHERE username = $1 +SELECT id, name, avatar_hash, created_at, username, password_hash, email, record_listens FROM members WHERE username = $1 ` func (q *Queries) AccountByUsername(ctx context.Context, username *string) (Member, error) { @@ -24,13 +24,14 @@ func (q *Queries) AccountByUsername(ctx context.Context, username *string) (Memb &i.Username, &i.PasswordHash, &i.Email, + &i.RecordListens, ) return i, err } const createMember = `-- name: CreateMember :one INSERT INTO members (name) VALUES ($1) -RETURNING id, name, avatar_hash, created_at, username, password_hash, email +RETURNING id, name, avatar_hash, created_at, username, password_hash, email, record_listens ` func (q *Queries) CreateMember(ctx context.Context, name string) (Member, error) { @@ -44,6 +45,7 @@ func (q *Queries) CreateMember(ctx context.Context, name string) (Member, error) &i.Username, &i.PasswordHash, &i.Email, + &i.RecordListens, ) return i, err } @@ -122,6 +124,7 @@ SELECT coalesce(password_hash, '')::text AS password_hash FROM members WHERE id ` // 让修改密码等到这次登录提交之后，它随后删除会话时会删掉这次登录签发的会话。 +// 用互斥的 FOR NO KEY UPDATE，不用 FOR SHARE：同一个账号的几次登录依次并入游客的历史，否则两边会按相反的顺序抢同一批历史行而死锁。 func (q *Queries) LockMemberPassword(ctx context.Context, id string) (string, error) { row := q.db.QueryRow(ctx, lockMemberPassword, id) var password_hash string @@ -146,7 +149,7 @@ func (q *Queries) MemberAccountProfile(ctx context.Context, id string) (MemberAc } const memberByToken = `-- name: MemberByToken :one -SELECT id, name, avatar_hash, created_at, username, password_hash, email FROM members WHERE id = (SELECT member_id FROM member_sessions WHERE token_hash = $1) +SELECT id, name, avatar_hash, created_at, username, password_hash, email, record_listens FROM members WHERE id = (SELECT member_id FROM member_sessions WHERE token_hash = $1) ` func (q *Queries) MemberByToken(ctx context.Context, tokenHash []byte) (Member, error) { @@ -160,6 +163,7 @@ func (q *Queries) MemberByToken(ctx context.Context, tokenHash []byte) (Member, &i.Username, &i.PasswordHash, &i.Email, + &i.RecordListens, ) return i, err } @@ -217,13 +221,24 @@ func (q *Queries) SetMemberPassword(ctx context.Context, arg SetMemberPasswordPa return result.RowsAffected(), nil } +const takeSession = `-- name: TakeSession :one +DELETE FROM member_sessions WHERE token_hash = $1 RETURNING member_id +` + +func (q *Queries) TakeSession(ctx context.Context, tokenHash []byte) (string, error) { + row := q.db.QueryRow(ctx, takeSession, tokenHash) + var member_id string + err := row.Scan(&member_id) + return member_id, err +} + const updateMember = `-- name: UpdateMember :one UPDATE members SET name = coalesce($1, name), avatar_hash = CASE WHEN $2::text IS NULL THEN avatar_hash ELSE NULLIF($2::text, '') END, email = CASE WHEN username IS NULL THEN NULL WHEN $3::text IS NULL THEN email ELSE NULLIF($3::text, '') END WHERE id = $4 -RETURNING id, name, avatar_hash, created_at, username, password_hash, email +RETURNING id, name, avatar_hash, created_at, username, password_hash, email, record_listens ` type UpdateMemberParams struct { @@ -250,6 +265,7 @@ func (q *Queries) UpdateMember(ctx context.Context, arg UpdateMemberParams) (Mem &i.Username, &i.PasswordHash, &i.Email, + &i.RecordListens, ) return i, err } diff --git a/internal/adapter/out/postgres/internal/query/models.go b/internal/adapter/out/postgres/internal/query/models.go index 7097cd8..dbf9fd7 100644 --- a/internal/adapter/out/postgres/internal/query/models.go +++ b/internal/adapter/out/postgres/internal/query/models.go @@ -9,13 +9,14 @@ import ( ) type Member struct { - ID string - Name string - AvatarHash *string - CreatedAt time.Time - Username *string - PasswordHash *string - Email *string + ID string + Name string + AvatarHash *string + CreatedAt time.Time + Username *string + PasswordHash *string + Email *string + RecordListens bool } type PlaylistItem struct { diff --git a/internal/adapter/out/postgres/listens.go b/internal/adapter/out/postgres/listens.go new file mode 100644 index 0000000..2375c66 --- /dev/null +++ b/internal/adapter/out/postgres/listens.go @@ -0,0 +1,119 @@ +package postgres + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + + "github.com/dihemoe/alisten/internal/adapter/out/postgres/internal/query" + "github.com/dihemoe/alisten/internal/app/port" + "github.com/dihemoe/alisten/internal/domain/track" +) + +var ( + _ port.ListenRecorder = (*Store)(nil) + _ port.Listens = (*Store)(nil) +) + +func (s *Store) RecordListen(ctx context.Context, memberIDs []string, l port.Listen) error { + err := pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error { + q := s.q.WithTx(tx) + ids, err := q.LockListeners(ctx, memberIDs) + if err != nil || len(ids) == 0 { + return err + } + m := l.Meta + err = q.UpsertListens(ctx, query.UpsertListensParams{ + MemberIds: ids, + Source: string(l.Track.Source), + TrackID: l.Track.ID, + Title: m.Title, + Artist: nullable(m.Artist), + Album: nullable(m.Album), + CoverURL: nullable(m.CoverURL), + CoverKey: nullable(string(m.CoverKey)), + DurationMs: m.Duration.Milliseconds(), + PageURL: m.PageURL, + HeardAt: l.HeardAt, + }) + if err != nil { + return err + } + return q.TrimListens(ctx, query.TrimListensParams{MemberIds: ids, Keep: port.ListenLimit}) + }) + if err != nil { + return fmt.Errorf("给 %d 个成员记录听到的 %s:%s：%w", len(memberIDs), l.Track.Source, l.Track.ID, err) + } + return nil +} + +// 在替换会话的事务中调用：账号行已经被互斥地锁住，与同时进行的记录、同时登录这个账号的其他游客依次写入账号的历史。 +func moveListens(ctx context.Context, q *query.Queries, fromID, toID string) error { + if err := q.MoveListens(ctx, query.MoveListensParams{FromID: fromID, ToID: toID}); err != nil { + return err + } + return q.TrimListens(ctx, query.TrimListensParams{MemberIds: []string{toID}, Keep: port.ListenLimit}) +} + +func (s *Store) MemberListens(ctx context.Context, memberID string) ([]port.Listen, error) { + rows, err := s.q.MemberListens(ctx, query.MemberListensParams{MemberID: memberID, Max: port.ListenLimit}) + if err != nil { + return nil, fmt.Errorf("读取成员 %s 的播放历史：%w", memberID, err) + } + listens := make([]port.Listen, len(rows)) + for i, r := range rows { + listens[i] = port.Listen{ + Track: track.Track{Source: track.Source(r.Source), ID: r.TrackID}, + Meta: track.Meta{ + Title: r.Title, + Artist: orEmpty(r.Artist), + Album: orEmpty(r.Album), + CoverURL: orEmpty(r.CoverURL), + CoverKey: track.ObjectKey(orEmpty(r.CoverKey)), + PageURL: r.PageURL, + Duration: time.Duration(r.DurationMs) * time.Millisecond, + }, + HeardAt: r.HeardAt, + } + } + return listens, nil +} + +func (s *Store) DeleteListen(ctx context.Context, memberID string, t track.Track) error { + if err := s.q.DeleteListen(ctx, query.DeleteListenParams{MemberID: memberID, Source: string(t.Source), TrackID: t.ID}); err != nil { + return fmt.Errorf("删除成员 %s 听过的 %s:%s：%w", memberID, t.Source, t.ID, err) + } + return nil +} + +func (s *Store) ClearListens(ctx context.Context, memberID string) error { + if err := s.q.ClearListens(ctx, memberID); err != nil { + return fmt.Errorf("清空成员 %s 的播放历史：%w", memberID, err) + } + return nil +} + +func (s *Store) ListenRecording(ctx context.Context, memberID string) (bool, error) { + on, err := s.q.ListenRecording(ctx, memberID) + if errors.Is(err, pgx.ErrNoRows) { + err = port.ErrMemberNotFound + } + if err != nil { + return false, fmt.Errorf("读取成员 %s 是否记录播放历史：%w", memberID, err) + } + return on, nil +} + +func (s *Store) SetListenRecording(ctx context.Context, memberID string, on bool) error { + n, err := s.q.SetListenRecording(ctx, query.SetListenRecordingParams{ID: memberID, RecordListens: on}) + if err == nil && n == 0 { + err = port.ErrMemberNotFound + } + if err != nil { + return fmt.Errorf("设置成员 %s 是否记录播放历史：%w", memberID, err) + } + return nil +} diff --git a/internal/adapter/out/postgres/listens_test.go b/internal/adapter/out/postgres/listens_test.go new file mode 100644 index 0000000..0129387 --- /dev/null +++ b/internal/adapter/out/postgres/listens_test.go @@ -0,0 +1,346 @@ +package postgres + +import ( + "errors" + "fmt" + "slices" + "sync" + "testing" + "time" + + "github.com/dihemoe/alisten/internal/app/port" + "github.com/dihemoe/alisten/internal/domain/track" +) + +var heard = time.Date(2026, 11, 10, 20, 0, 0, 0, time.UTC) + +func sunnyAt(at time.Time) port.Listen { + return port.Listen{ + Track: track.Track{Source: track.NetEase, ID: "186016"}, + Meta: track.Meta{Title: "晴天", Artist: "周杰伦", Album: "叶惠美", CoverURL: "https://p1.music.126.net/a.jpg", PageURL: "https://music.163.com/#/song?id=186016", Duration: 269500 * time.Millisecond}, + HeardAt: at, + } +} + +func clipAt(at time.Time) port.Listen { + return port.Listen{ + Track: track.Track{Source: track.Link, ID: "https://example.com/a.mp3"}, + Meta: track.Meta{Title: "a.mp3", CoverKey: "covers/abc", PageURL: "https://example.com/a.mp3", Duration: time.Minute}, + HeardAt: at, + } +} + +func songAt(id string, at time.Time) port.Listen { + return port.Listen{Track: track.Track{Source: track.NetEase, ID: id}, Meta: track.Meta{Title: id, PageURL: track.NetEasePageURL(id)}, HeardAt: at} +} + +func record(t *testing.T, store *Store, members []string, l port.Listen) { + t.Helper() + if err := store.RecordListen(t.Context(), members, l); err != nil { + t.Fatal(err) + } +} + +// 时间统一换成 UTC，便于直接比较。 +func listensOf(t *testing.T, store *Store, memberID string) []port.Listen { + t.Helper() + listens, err := store.MemberListens(t.Context(), memberID) + if err != nil { + t.Fatal(err) + } + for i := range listens { + listens[i].HeardAt = listens[i].HeardAt.UTC() + } + return listens +} + +func heardIDs(listens []port.Listen) []string { + ids := make([]string, len(listens)) + for i, l := range listens { + ids[i] = l.Track.ID + } + return ids +} + +func TestRecordListenKeepsLatestHearingOfEachTrack(t *testing.T) { + store, _ := newStore(t) + alice := createMember(t, store, "alice", "游客-1a2b") + bob := createMember(t, store, "bob", "游客-3c4d") + record(t, store, []string{alice.ID, bob.ID}, sunnyAt(heard)) + record(t, store, []string{alice.ID}, clipAt(heard.Add(time.Minute))) + if got, want := listensOf(t, store, alice.ID), []port.Listen{clipAt(heard.Add(time.Minute)), sunnyAt(heard)}; !slices.Equal(got, want) { + t.Errorf("alice 的历史 = %#v，期望最近听到的在前 %#v", got, want) + } + if got, want := listensOf(t, store, bob.ID), []port.Listen{sunnyAt(heard)}; !slices.Equal(got, want) { + t.Errorf("bob 的历史 = %#v，期望 %#v", got, want) + } + + // 同一首歌再次记录：heard_at 取较晚的，元数据用新的。 + renamed := sunnyAt(heard.Add(-time.Hour)) + renamed.Meta.Title = "晴天（重制）" + record(t, store, []string{alice.ID}, renamed) + want := sunnyAt(heard) + want.Meta.Title = "晴天（重制）" + if got := listensOf(t, store, alice.ID); !slices.Equal(got, []port.Listen{clipAt(heard.Add(time.Minute)), want}) { + t.Errorf("较早的一次记录之后 = %#v，期望保留较晚的时间、换成新的元数据", got) + } + record(t, store, []string{alice.ID}, sunnyAt(heard.Add(time.Hour))) + if got := heardIDs(listensOf(t, store, alice.ID)); !slices.Equal(got, []string{"186016", "https://example.com/a.mp3"}) { + t.Errorf("再次听到《晴天》之后的顺序 = %v，期望它排到最前", got) + } +} + +func TestRecordListenTruncatesToStoredPrecision(t *testing.T) { + store, _ := newStore(t) + alice := createMember(t, store, "alice", "游客-1a2b") + l := sunnyAt(heard.Add(1234567 * time.Nanosecond)) + l.Meta.Duration = 3*time.Minute + 1234*time.Microsecond + record(t, store, []string{alice.ID}, l) + got := listensOf(t, store, alice.ID) + if len(got) != 1 || got[0].Meta.Duration != 3*time.Minute+time.Millisecond || !got[0].HeardAt.Equal(heard.Add(1234*time.Microsecond)) { + t.Errorf("历史 = %#v，期望时长按毫秒、时间按微秒截断", got) + } +} + +func TestRecordListenSkipsMembersWhoTurnedRecordingOff(t *testing.T) { + store, _ := newStore(t) + alice := createMember(t, store, "alice", "游客-1a2b") + bob := createMember(t, store, "bob", "游客-3c4d") + if on, err := store.ListenRecording(t.Context(), bob.ID); err != nil || !on { + t.Fatalf("新成员的 ListenRecording = %v, %v，期望默认开启", on, err) + } + record(t, store, []string{bob.ID}, sunnyAt(heard)) + if err := store.SetListenRecording(t.Context(), bob.ID, false); err != nil { + t.Fatal(err) + } + if on, err := store.ListenRecording(t.Context(), bob.ID); err != nil || on { + t.Errorf("关闭之后 ListenRecording = %v, %v", on, err) + } + record(t, store, []string{alice.ID, bob.ID, "01a15778-1911-7a28-8b2e-e08c2a25f388"}, clipAt(heard.Add(time.Minute))) + if got := heardIDs(listensOf(t, store, alice.ID)); !slices.Equal(got, []string{"https://example.com/a.mp3"}) { + t.Errorf("alice 的历史 = %v", got) + } + if got := heardIDs(listensOf(t, store, bob.ID)); !slices.Equal(got, []string{"186016"}) { + t.Errorf("bob 关闭记录之后的历史 = %v，期望只有关闭之前的", got) + } + if err := store.SetListenRecording(t.Context(), bob.ID, true); err != nil { + t.Fatal(err) + } + record(t, store, []string{bob.ID}, clipAt(heard.Add(2*time.Minute))) + if got := heardIDs(listensOf(t, store, bob.ID)); len(got) != 2 { + t.Errorf("重新开启之后的历史 = %v，期望继续记录", got) + } + + missing := "01a15778-1911-7a28-8b2e-e08c2a25f388" + if _, err := store.ListenRecording(t.Context(), missing); !errors.Is(err, port.ErrMemberNotFound) { + t.Errorf("不存在的成员的 ListenRecording 错误 = %v，期望 ErrMemberNotFound", err) + } + if err := store.SetListenRecording(t.Context(), missing, false); !errors.Is(err, port.ErrMemberNotFound) { + t.Errorf("不存在的成员的 SetListenRecording 错误 = %v，期望 ErrMemberNotFound", err) + } +} + +func TestRecordListenKeepsLatest500(t *testing.T) { + store, pool := newStore(t) + alice := createMember(t, store, "alice", "游客-1a2b") + bob := createMember(t, store, "bob", "游客-3c4d") + // 第 i 首在 heard 之后 i 分钟听到，s0 最早。 + for _, m := range []string{alice.ID, bob.ID} { + if _, err := pool.Exec(t.Context(), ` + INSERT INTO listens (member_id, source, track_id, title, duration_ms, page_url, heard_at) + SELECT $1, 'netease', 's' || i, 's' || i, 0, 'https://music.163.com', $2::timestamptz + i * interval '1 minute' + FROM generate_series(0, 499) AS i`, m, heard); err != nil { + t.Fatal(err) + } + } + record(t, store, []string{alice.ID, bob.ID}, songAt("new", heard.Add(time.Hour*24))) + for _, m := range []string{alice.ID, bob.ID} { + got := heardIDs(listensOf(t, store, m)) + if len(got) != port.ListenLimit || got[0] != "new" || got[len(got)-1] != "s1" { + t.Errorf("第 501 首之后有 %d 首，最新 %q、最早 %q，期望 500 首，s0 被删掉", len(got), got[0], got[len(got)-1]) + } + } + // 比现有的都早的一首，记下之后它自己就是超出的部分。 + record(t, store, []string{alice.ID}, songAt("old", heard.Add(-time.Hour))) + if got := heardIDs(listensOf(t, store, alice.ID)); slices.Contains(got, "old") || len(got) != port.ListenLimit { + t.Errorf("更早的一首记下之后有 %d 首，含它 = %v，期望不留", len(got), slices.Contains(got, "old")) + } +} + +// 同时开播的几个房间里有同样的两个成员，各自按不同的顺序给出成员，并发写入都成功，每首歌只留一条。 +func TestConcurrentRecordsOfSameMembers(t *testing.T) { + store, _ := newStore(t) + alice := createMember(t, store, "alice", "游客-1a2b") + bob := createMember(t, store, "bob", "游客-3c4d") + var wg sync.WaitGroup + errs := make(chan error, 40) + for i := range 40 { + members := []string{alice.ID, bob.ID} + if i%2 == 1 { + members = []string{bob.ID, alice.ID} + } + wg.Go(func() { + errs <- store.RecordListen(t.Context(), members, songAt(fmt.Sprint(i/2%4), heard.Add(time.Duration(i)*time.Second))) + }) + } + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + for _, m := range []string{alice.ID, bob.ID} { + if got := heardIDs(listensOf(t, store, m)); !slices.Equal(got, []string{"3", "2", "1", "0"}) { + t.Errorf("并发记录之后的历史 = %v，期望每首一条、按最后一次听到的时间排列", got) + } + } +} + +func TestDeleteAndClearListens(t *testing.T) { + store, pool := newStore(t) + alice := createMember(t, store, "alice", "游客-1a2b") + bob := createMember(t, store, "bob", "游客-3c4d") + record(t, store, []string{alice.ID, bob.ID}, sunnyAt(heard)) + record(t, store, []string{alice.ID, bob.ID}, clipAt(heard.Add(time.Minute))) + + if err := store.DeleteListen(t.Context(), alice.ID, sunnyAt(heard).Track); err != nil { + t.Fatal(err) + } + if err := store.DeleteListen(t.Context(), alice.ID, track.Track{Source: track.QQ, ID: "missing"}); err != nil { + t.Errorf("删除不在历史里的歌 = %v，期望忽略", err) + } + if got := heardIDs(listensOf(t, store, alice.ID)); !slices.Equal(got, []string{"https://example.com/a.mp3"}) { + t.Errorf("删除《晴天》之后 alice 的历史 = %v", got) + } + if err := store.ClearListens(t.Context(), alice.ID); err != nil { + t.Fatal(err) + } + if got := listensOf(t, store, alice.ID); len(got) != 0 { + t.Errorf("清空之后 alice 的历史 = %v", got) + } + if got := heardIDs(listensOf(t, store, bob.ID)); len(got) != 2 { + t.Errorf("alice 删除和清空之后 bob 的历史 = %v，期望不受影响", got) + } + + if _, err := pool.Exec(t.Context(), "DELETE FROM members WHERE id = $1", bob.ID); err != nil { + t.Fatal(err) + } + var left int + if err := pool.QueryRow(t.Context(), "SELECT count(*) FROM listens WHERE member_id = $1", bob.ID).Scan(&left); err != nil || left != 0 { + t.Errorf("删除成员之后还剩 %d 条历史（%v），期望随成员删除", left, err) + } +} + +func TestReplaceSessionMovesGuestListensIntoAccount(t *testing.T) { + store, _ := newStore(t) + alice := registered(t, store, "alice") + guest := createMember(t, store, "guest", "游客-3c4d") + sunny := sunnyAt(heard.Add(time.Hour)) + sunny.Meta.Title = "晴天（游客听到的）" + record(t, store, []string{alice.ID}, sunnyAt(heard)) + record(t, store, []string{alice.ID}, songAt("1", heard.Add(time.Minute))) + record(t, store, []string{guest.ID}, sunny) + record(t, store, []string{guest.ID}, clipAt(heard.Add(-time.Hour))) + + if err := store.ReplaceSession(t.Context(), tokenHash("guest"), alice.ID, tokenHash("alice-2"), "hash"); err != nil { + t.Fatal(err) + } + if got, want := listensOf(t, store, alice.ID), []port.Listen{sunny, songAt("1", heard.Add(time.Minute)), clipAt(heard.Add(-time.Hour))}; !slices.Equal(got, want) { + t.Errorf("登录之后账号的历史 = %#v，期望并入游客的历史 %#v", got, want) + } + if got := listensOf(t, store, guest.ID); len(got) != 0 { + t.Errorf("登录之后游客的历史 = %v，期望移走", got) + } +} + +func TestReplaceSessionMergeFollowsRecordingRules(t *testing.T) { + store, pool := newStore(t) + alice := registered(t, store, "alice") + guest := createMember(t, store, "guest", "游客-3c4d") + other := createMember(t, store, "other", "游客-5e6f") + if _, err := pool.Exec(t.Context(), ` + INSERT INTO listens (member_id, source, track_id, title, duration_ms, page_url, heard_at) + SELECT $1, 'netease', 's' || i, 's' || i, 0, 'https://music.163.com', $2::timestamptz + i * interval '1 minute' + FROM generate_series(0, 499) AS i`, alice.ID, heard); err != nil { + t.Fatal(err) + } + record(t, store, []string{guest.ID}, songAt("new", heard.Add(24*time.Hour))) + if err := store.ReplaceSession(t.Context(), tokenHash("guest"), alice.ID, tokenHash("alice-2"), "hash"); err != nil { + t.Fatal(err) + } + if got := heardIDs(listensOf(t, store, alice.ID)); len(got) != port.ListenLimit || got[0] != "new" || slices.Contains(got, "s0") { + t.Errorf("并入之后账号有 %d 首，期望 500 首、最早的 s0 被删掉", len(got)) + } + + if err := store.SetListenRecording(t.Context(), alice.ID, false); err != nil { + t.Fatal(err) + } + record(t, store, []string{other.ID}, songAt("other", heard.Add(48*time.Hour))) + if err := store.ReplaceSession(t.Context(), tokenHash("other"), alice.ID, tokenHash("alice-3"), "hash"); err != nil { + t.Fatal(err) + } + if got := heardIDs(listensOf(t, store, alice.ID)); slices.Contains(got, "other") { + t.Error("账号关闭记录时不应并入游客的历史") + } + if got := listensOf(t, store, other.ID); len(got) != 0 { + t.Errorf("游客的历史 = %v，账号不并入时也随会话一起删掉", got) + } +} + +func TestReplaceSessionKeepsListensWhenRejected(t *testing.T) { + store, _ := newStore(t) + alice := registered(t, store, "alice") + guest := createMember(t, store, "guest", "游客-3c4d") + record(t, store, []string{guest.ID}, sunnyAt(heard)) + if err := store.ReplaceSession(t.Context(), tokenHash("guest"), alice.ID, tokenHash("alice-2"), "changed"); !errors.Is(err, port.ErrAccountPasswordChanged) { + t.Fatalf("密码哈希不符时 = %v", err) + } + if err := store.ReplaceSession(t.Context(), tokenHash("nobody"), alice.ID, tokenHash("alice-2"), "hash"); !errors.Is(err, port.ErrSessionNotFound) { + t.Fatalf("旧会话不存在时 = %v", err) + } + if got := listensOf(t, store, guest.ID); len(got) != 1 { + t.Errorf("登录被拒绝之后游客的历史 = %v，期望不变", got) + } + if got := listensOf(t, store, alice.ID); len(got) != 0 { + t.Errorf("登录被拒绝之后账号的历史 = %v，期望不变", got) + } +} + +// 两台设备上的游客同时登录同一个账号，两边听过同一批歌、写入的先后相反：两次合并依次进行，都成功，账号里每首歌只留一条。 +func TestConcurrentReplaceSessionsMergeIntoSameAccount(t *testing.T) { + store, pool := newStore(t) + const songs = 200 + for round := range 5 { + alice := registered(t, store, fmt.Sprintf("alice-%d", round)) + guests := []string{fmt.Sprintf("guest-%d-a", round), fmt.Sprintf("guest-%d-b", round)} + for i, token := range guests { + guest := createMember(t, store, token, "游客-"+token) + if _, err := pool.Exec(t.Context(), ` + INSERT INTO listens (member_id, source, track_id, title, duration_ms, page_url, heard_at) + SELECT $1, 'netease', 's' || i, 's' || i, 0, 'https://music.163.com', $2::timestamptz + i * interval '1 minute' + FROM generate_series(0, $3 - 1) AS i ORDER BY i * $4`, guest.ID, heard, songs, 1-2*i); err != nil { + t.Fatal(err) + } + } + + start := make(chan struct{}) + errs := make([]error, len(guests)) + var wg sync.WaitGroup + for i, token := range guests { + wg.Go(func() { + <-start + errs[i] = store.ReplaceSession(t.Context(), tokenHash(token), alice.ID, tokenHash(token+"-alice"), "hash") + }) + } + close(start) + wg.Wait() + if err := errors.Join(errs...); err != nil { + t.Fatalf("第 %d 轮同时登录：%v", round, err) + } + got := heardIDs(listensOf(t, store, alice.ID)) + if len(got) != songs || got[0] != fmt.Sprintf("s%d", songs-1) || got[songs-1] != "s0" { + t.Errorf("第 %d 轮并入之后账号有 %d 首，期望 %d 首、每首一条、按听到的时间排列", round, len(got), songs) + } + } +} diff --git a/internal/adapter/out/postgres/members.go b/internal/adapter/out/postgres/members.go index 56ffe74..705162c 100644 --- a/internal/adapter/out/postgres/members.go +++ b/internal/adapter/out/postgres/members.go @@ -116,14 +116,20 @@ func (s *Store) ReplaceSession(ctx context.Context, oldTokenHash []byte, memberI if current != passwordHash { return port.ErrAccountPasswordChanged } - n, err := q.DeleteSession(ctx, oldTokenHash) + previous, err := q.TakeSession(ctx, oldTokenHash) + if errors.Is(err, pgx.ErrNoRows) { + return port.ErrSessionNotFound + } if err != nil { return err } - if n == 0 { - return port.ErrSessionNotFound + if err := q.CreateSession(ctx, query.CreateSessionParams{TokenHash: newTokenHash, MemberID: memberID}); err != nil { + return err } - return q.CreateSession(ctx, query.CreateSessionParams{TokenHash: newTokenHash, MemberID: memberID}) + if previous == memberID { + return nil + } + return moveListens(ctx, q, previous, memberID) }) if err != nil { return fmt.Errorf("给成员 %s 更换会话：%w", memberID, err) diff --git a/internal/app/identity/account_test.go b/internal/app/identity/account_test.go index d38967e..2ac5a98 100644 --- a/internal/app/identity/account_test.go +++ b/internal/app/identity/account_test.go @@ -15,6 +15,7 @@ import ( "github.com/dihemoe/alisten/internal/app/port" "github.com/dihemoe/alisten/internal/app/port/porttest" "github.com/dihemoe/alisten/internal/domain/member" + "github.com/dihemoe/alisten/internal/domain/track" ) const clientIP = "203.0.113.7" @@ -252,6 +253,30 @@ func TestLogin(t *testing.T) { } } +func TestLoginMovesGuestListensIntoAccount(t *testing.T) { + a, s, store, _, _ := newAccounts(t) + alice := registerAccount(t, a, s, "alice", "correct horse") + guest := createMember(t, s) + heard := time.Date(2026, 11, 10, 20, 0, 0, 0, time.UTC) + sunny := port.Listen{Track: track.Track{Source: track.NetEase, ID: "186016"}, Meta: track.Meta{Title: "晴天"}, HeardAt: heard} + rain := port.Listen{Track: track.Track{Source: track.NetEase, ID: "185811"}, Meta: track.Meta{Title: "七里香"}, HeardAt: heard.Add(time.Minute)} + if err := store.RecordListen(t.Context(), []string{alice.Member.ID}, sunny); err != nil { + t.Fatal(err) + } + if err := store.RecordListen(t.Context(), []string{guest.Member.ID}, rain); err != nil { + t.Fatal(err) + } + if _, _, _, err := a.Login(t.Context(), guest.Member, guest.MemberToken, clientIP, "alice", "correct horse", ""); err != nil { + t.Fatal(err) + } + if listens, err := store.MemberListens(t.Context(), alice.Member.ID); err != nil || !slices.Equal(listens, []port.Listen{rain, sunny}) { + t.Errorf("登录之后账号的历史 = %#v, %v，期望并入游客听过的《七里香》", listens, err) + } + if listens, err := store.MemberListens(t.Context(), guest.Member.ID); err != nil || len(listens) != 0 { + t.Errorf("登录之后游客的历史 = %#v, %v，期望移走", listens, err) + } +} + func TestLoginRejected(t *testing.T) { a, s, store, q, _ := newAccounts(t) alice := registerAccount(t, a, s, "alice", "correct horse") diff --git a/internal/app/port/listen.go b/internal/app/port/listen.go new file mode 100644 index 0000000..7304135 --- /dev/null +++ b/internal/app/port/listen.go @@ -0,0 +1,36 @@ +package port + +import ( + "context" + "time" + + "github.com/dihemoe/alisten/internal/domain/track" +) + +// 每个成员的播放历史最多保留的歌数，超出的部分从最早听到的删起。 +const ListenLimit = 500 + +// 成员在房间里听到过的一首歌。元数据是听到时的快照。 +type Listen struct { + Track track.Track + Meta track.Meta + HeardAt time.Time +} + +type ListenRecorder interface { + // 只记给开启了记录的成员。同一首歌已有记录时 HeardAt 取较晚的，元数据用新的； + // 每个成员超过 ListenLimit 首的部分随即删掉。 + RecordListen(ctx context.Context, memberIDs []string, l Listen) error +} + +type Listens interface { + // 最近听到的在前，最多 ListenLimit 首。 + MemberListens(ctx context.Context, memberID string) ([]Listen, error) + // 不在历史里的歌忽略。 + DeleteListen(ctx context.Context, memberID string, t track.Track) error + ClearListens(ctx context.Context, memberID string) error + // 成员不存在时返回 ErrMemberNotFound。 + ListenRecording(ctx context.Context, memberID string) (bool, error) + // 关闭之后已有的历史保留。成员不存在时返回 ErrMemberNotFound。 + SetListenRecording(ctx context.Context, memberID string, on bool) error +} diff --git a/internal/app/port/member.go b/internal/app/port/member.go index 877083f..0804c87 100644 --- a/internal/app/port/member.go +++ b/internal/app/port/member.go @@ -44,6 +44,8 @@ type Members interface { // 在同一个事务中删除 oldTokenHash 的会话，给 memberID 创建 newTokenHash 的会话。oldTokenHash 的会话已经不存在时返回 // ErrSessionNotFound；memberID 的密码哈希已经不是 passwordHash 时返回 ErrAccountPasswordChanged：密码在校验之后被修改过， // 修改时对会话的删除已经执行，此时创建的会话不会再被删除。两种情况都不做任何修改。 + // 同一个事务中把 oldTokenHash 所属成员的播放历史移到 memberID，并入的规则与 ListenRecorder.RecordListen 相同， + // memberID 关闭了记录时不并入；两者是同一个成员时不动。 ReplaceSession(ctx context.Context, oldTokenHash []byte, memberID string, newTokenHash []byte, passwordHash string) error DeleteSession(ctx context.Context, tokenHash []byte) error // 在同一个事务中修改已注册成员的密码哈希，并删除该成员的全部会话。返回被删除的会话的 token 哈希。成员是游客时返回 ErrMemberNotFound。 diff --git a/internal/app/port/porttest/listen.go b/internal/app/port/porttest/listen.go new file mode 100644 index 0000000..6e0ba58 --- /dev/null +++ b/internal/app/port/porttest/listen.go @@ -0,0 +1,91 @@ +package porttest + +import ( + "cmp" + "context" + "slices" + "time" + + "github.com/dihemoe/alisten/internal/app/port" + "github.com/dihemoe/alisten/internal/domain/track" +) + +var ( + _ port.ListenRecorder = (*Store)(nil) + _ port.Listens = (*Store)(nil) +) + +func (s *Store) RecordListen(_ context.Context, memberIDs []string, l port.Listen) error { + m := l.Meta + if err := checkText(string(l.Track.Source), l.Track.ID, m.Title, m.Artist, m.Album, m.CoverURL, string(m.CoverKey), m.PageURL); err != nil { + return err + } + // 与 PostgreSQL 存下的精度一致。 + l.Meta.Duration = m.Duration.Truncate(time.Millisecond) + l.HeardAt = l.HeardAt.Truncate(time.Microsecond) + s.mu.Lock() + defer s.mu.Unlock() + for _, id := range memberIDs { + if stored := s.member(id); stored != nil && !stored.listensOff { + s.addListen(id, l) + } + } + return nil +} + +// 调用方持有 s.mu。 +func (s *Store) addListen(memberID string, l port.Listen) { + list := s.listens[memberID] + if i := slices.IndexFunc(list, func(old port.Listen) bool { return old.Track == l.Track }); i >= 0 { + if list[i].HeardAt.After(l.HeardAt) { + l.HeardAt = list[i].HeardAt + } + list = slices.Delete(list, i, i+1) + } + list = append(list, l) + slices.SortFunc(list, func(a, b port.Listen) int { + return cmp.Or(b.HeardAt.Compare(a.HeardAt), cmp.Compare(a.Track.Source, b.Track.Source), cmp.Compare(a.Track.ID, b.Track.ID)) + }) + s.listens[memberID] = list[:min(len(list), port.ListenLimit)] +} + +func (s *Store) MemberListens(_ context.Context, memberID string) ([]port.Listen, error) { + s.mu.Lock() + defer s.mu.Unlock() + return slices.Clone(s.listens[memberID]), nil +} + +func (s *Store) DeleteListen(_ context.Context, memberID string, t track.Track) error { + s.mu.Lock() + defer s.mu.Unlock() + s.listens[memberID] = slices.DeleteFunc(s.listens[memberID], func(l port.Listen) bool { return l.Track == t }) + return nil +} + +func (s *Store) ClearListens(_ context.Context, memberID string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.listens, memberID) + return nil +} + +func (s *Store) ListenRecording(_ context.Context, memberID string) (bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + m := s.member(memberID) + if m == nil { + return false, port.ErrMemberNotFound + } + return !m.listensOff, nil +} + +func (s *Store) SetListenRecording(_ context.Context, memberID string, on bool) error { + s.mu.Lock() + defer s.mu.Unlock() + m := s.member(memberID) + if m == nil { + return port.ErrMemberNotFound + } + m.listensOff = !on + return nil +} diff --git a/internal/app/port/porttest/store.go b/internal/app/port/porttest/store.go index 7deb567..c70ce68 100644 --- a/internal/app/port/porttest/store.go +++ b/internal/app/port/porttest/store.go @@ -36,6 +36,8 @@ type Store struct { brandName string registrationClosed bool playlists []*storedPlaylist + // 成员 ID 到播放历史，按 MemberListens 的顺序排列。 + listens map[string][]port.Listen } type storedMember struct { @@ -43,6 +45,8 @@ type storedMember struct { username string email string passwordHash string + // 零值表示记录播放历史，与 members.record_listens 的默认值一致。 + listensOff bool } type storedTicket struct { @@ -70,6 +74,7 @@ func New() *Store { tickets: map[string]storedTicket{}, bots: map[string]string{}, cookies: map[string]string{}, + listens: map[string][]port.Listen{}, } } @@ -205,7 +210,8 @@ func (s *Store) ReplaceSession(_ context.Context, oldTokenHash []byte, memberID if m.passwordHash != passwordHash { return port.ErrAccountPasswordChanged } - if _, ok := s.sessions[string(oldTokenHash)]; !ok { + previous, ok := s.sessions[string(oldTokenHash)] + if !ok { return port.ErrSessionNotFound } if _, ok := s.sessions[string(newTokenHash)]; ok { @@ -213,6 +219,15 @@ func (s *Store) ReplaceSession(_ context.Context, oldTokenHash []byte, memberID } s.deleteSession(string(oldTokenHash)) s.sessions[string(newTokenHash)] = memberID + if previous != memberID { + moved := s.listens[previous] + delete(s.listens, previous) + if !m.listensOff { + for _, l := range moved { + s.addListen(memberID, l) + } + } + } return nil } diff --git a/internal/app/port/porttest/store_test.go b/internal/app/port/porttest/store_test.go index 6d6ae6d..1539826 100644 --- a/internal/app/port/porttest/store_test.go +++ b/internal/app/port/porttest/store_test.go @@ -46,6 +46,30 @@ func TestAddPlaylistItemKeepsMilliseconds(t *testing.T) { } } +func TestRecordListenKeepsWhatPostgreSQLKeeps(t *testing.T) { + store := New() + m, err := store.CreateMember(t.Context(), []byte("token"), "甲") + if err != nil { + t.Fatal(err) + } + heard := time.Date(2026, 11, 10, 20, 0, 0, 1234567, time.UTC) + l := port.Listen{Track: track.Track{Source: "netease", ID: "1"}, Meta: track.Meta{Title: "歌", Duration: 3*time.Minute + 1234*time.Microsecond}, HeardAt: heard} + if err := store.RecordListen(t.Context(), []string{m.ID}, l); err != nil { + t.Fatal(err) + } + listens, err := store.MemberListens(t.Context(), m.ID) + if err != nil { + t.Fatal(err) + } + if len(listens) != 1 || listens[0].Meta.Duration != 3*time.Minute+time.Millisecond || !listens[0].HeardAt.Equal(heard.Truncate(time.Microsecond)) { + t.Errorf("历史 = %#v，期望时长按毫秒、时间按微秒截断", listens) + } + l.Meta.Title = "坏\x00标题" + if err := store.RecordListen(t.Context(), []string{m.ID}, l); err == nil { + t.Error("含 NUL 的标题应报错") + } +} + func TestImportPlaylistRejectsWhatPostgreSQLRejects(t *testing.T) { store := New() m, err := store.CreateMember(t.Context(), []byte("token"), "甲")