feat: keep each member's listening history and move a guest's history into the account on login
This commit is contained in:
1 parent
a12ab44ec8
commit
ce5ce141f8
15 files changed
+1013
-16
No files matched your search
@@ -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);
|
||||
@@ -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;
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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。
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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"), "甲")
|
||||
|
||||
Reference in new issue
Block a user