feat: bind room tickets to member sessions and disconnect revoked sessions

internal/adapter/in/api/api.go

@@ -68,6 +68,8 @@ type call struct {
 	member member.Member
 	roomID string
 	ticket string
+	// 票据签发给的成员会话的 token 哈希，API Key 没有会话。
+	memberSession []byte
 	// 票据是常驻房间的 API Key。
 	bot bool
 	// 发出请求的客户端地址，按来源限流时使用。
@@ -211,7 +213,7 @@ func (s *server) serve(ctx context.Context, w http.ResponseWriter, r *http.Reque
 		if err != nil {
 			return ctx, nil, err
 		}
-		c.roomID, c.member, c.bot = holder.RoomID, holder.Member, holder.Bot
+		c.roomID, c.member, c.memberSession, c.bot = holder.RoomID, holder.Member, holder.SessionTokenHash, holder.Bot
 		ctx = logctx.WithMember(logctx.WithRoom(ctx, holder.RoomID), holder.Member.ID)
 	case protocol.AuthAdmin:
 		if err := s.Admin.Authorize(bearer(r)); err != nil {

internal/adapter/in/api/api_test.go

@@ -42,7 +42,7 @@ func services(store *porttest.Store, rooms port.RoomDirectory, adminToken string
 	l := lobby.New(rooms, node)
 	return Services{
 		Identity: identity.New(store, rooms, store, gateway),
-		Accounts: identity.NewAccounts(store, store, nil),
+		Accounts: identity.NewAccounts(store, store, nil, gateway, slog.New(slog.DiscardHandler)),
 		Lobby:    l,
 		Admin:    admin.New(adminToken, l, rooms, store, store, porttest.NewNetEaseQRLogin(), porttest.NewQQQRLogin(), porttest.NewBilibiliQRLogin(), porttest.NewYouTubeCookies(), gateway),
 		Gateway:  gateway,

internal/adapter/in/api/events.go

@@ -23,7 +23,7 @@ const (
 
 var streamCloses = map[realtime.CloseReason]streamClose{
 	realtime.CloseRestart:  {Code: protocol.CloseServiceRestart, Message: "服务端实例退出或订阅中断，请重新连接"},
-	realtime.CloseRevoked:  {Code: protocol.CloseUnauthorized, Message: "凭证已失效：房间设置或修改了密码，或者 API Key 已重置"},
+	realtime.CloseRevoked:  {Code: protocol.CloseUnauthorized, Message: "凭证已失效：房间设置或修改了密码，API Key 已重置，或者账号在这台设备上退出了登录、密码被修改或重置"},
 	realtime.CloseRoomGone: {Code: protocol.CloseNotFound, Message: "房间已关闭或被删除"},
 	realtime.CloseOverflow: {Code: protocol.CloseSendQueueFull, Message: "积压的事件过多，请重新连接"},
 	// 不是 1012：共用一把 Key 的两个 bot 看到它就该停下，而不是重连之后互相顶掉。
@@ -83,7 +83,7 @@ func (s *server) roomEvents(w http.ResponseWriter, r *http.Request) {
 func (s *server) observe(ctx context.Context, c call) (any, error) {
 	// 内存中只留凭证的哈希。
 	credential := sha256.Sum256([]byte(c.ticket))
-	o, err := s.Gateway.Observe(ctx, c.roomID, c.member, c.bot, string(credential[:]))
+	o, err := s.Gateway.Observe(ctx, c.roomID, c.member, c.memberSession, c.bot, string(credential[:]))
 	if err != nil {
 		return nil, err
 	}

internal/adapter/in/api/events_test.go

@@ -145,7 +145,7 @@ func TestRoomEvents(t *testing.T) {
 		if _, err := f.admin.ResetRoomAPIKey(t.Context(), hall); err != nil {
 			t.Fatal(err)
 		}
-		checkEvent(t, r, "event: close", `data: {"code":4401,"message":"凭证已失效：房间设置或修改了密码，或者 API Key 已重置"}`)
+		checkEvent(t, r, "event: close", `data: {"code":4401,"message":"凭证已失效：房间设置或修改了密码，API Key 已重置，或者账号在这台设备上退出了登录、密码被修改或重置"}`)
 		checkEnded(t, r)
 	})
 }

internal/adapter/in/api/handlers.go

@@ -98,7 +98,7 @@ func (s *server) createTicket(ctx context.Context, c call) (any, error) {
 	if err != nil {
 		return nil, err
 	}
-	ticket, err := s.Identity.IssueTicket(ctx, c.member.ID, id, deref(req.Password))
+	ticket, err := s.Identity.IssueTicket(ctx, c.member.ID, bearer(c.r), id, deref(req.Password))
 	if err != nil {
 		return nil, err
 	}

internal/adapter/in/api/room_test.go

@@ -232,7 +232,7 @@ func TestRoomRequestOfDisconnectedClientIsNotCounted(t *testing.T) {
 func TestRoomListingShowsOnlineCount(t *testing.T) {
 	f := newRoomFixture(t)
 	id, m, _ := f.enter(t)
-	session, err := f.gateway.Join(t.Context(), id, m)
+	session, err := f.gateway.Join(t.Context(), id, m, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -252,7 +252,7 @@ func TestProfileChangeReachesRooms(t *testing.T) {
 	session := decode[protocol.CreateMemberResponse](t, send(t, f.server, http.MethodPost, "/api/members", "{}"))
 	created := decode[protocol.CreateRoomResponse](t, send(t, f.server, http.MethodPost, "/api/rooms", `{"name":"临时","description":""}`))
 	m := member.Member{ID: session.Member.ID, Name: session.Member.Name}
-	s, err := f.gateway.Join(t.Context(), created.ID, m)
+	s, err := f.gateway.Join(t.Context(), created.ID, m, nil)
 	if err != nil {
 		t.Fatal(err)
 	}

internal/adapter/in/api/ws_test.go

@@ -191,7 +191,7 @@ func (f *wsFixture) enter(t *testing.T, roomID string) (string, member.Member, s
 			t.Fatal(err)
 		}
 	}
-	ticket, err := f.identity.IssueTicket(t.Context(), session.Member.ID, roomID, "")
+	ticket, err := f.identity.IssueTicket(t.Context(), session.Member.ID, session.MemberToken, roomID, "")
 	if err != nil {
 		t.Fatal(err)
 	}

internal/adapter/out/postgres/apikeys_test.go

@@ -2,6 +2,7 @@ package postgres
 
 import (
 	"errors"
+	"reflect"
 	"testing"
 	"time"
 
@@ -58,7 +59,7 @@ func TestResetRoomAPIKey(t *testing.T) {
 	if _, err := store.TicketHolder(t.Context(), tokenHash("k1")); !errors.Is(err, port.ErrTicketNotFound) {
 		t.Errorf("重置之后旧 Key 的错误 = %v，期望 ErrTicketNotFound", err)
 	}
-	if holder, err := store.TicketHolder(t.Context(), tokenHash("k2")); err != nil || holder != bot {
+	if holder, err := store.TicketHolder(t.Context(), tokenHash("k2")); err != nil || !reflect.DeepEqual(holder, bot) {
 		t.Errorf("重置之后新 Key 的持有者 = %#v, %v，期望 bot 成员不变", holder, err)
 	}
 	if n := memberTickets(t, pool, bot.Member.ID); n != 1 {
@@ -84,7 +85,7 @@ func TestSettingPasswordKeepsAPIKey(t *testing.T) {
 	if err := store.UpdateRoomAccess(t.Context(), id, port.AccessChange{PasswordHash: ptr("h1")}); err != nil {
 		t.Fatal(err)
 	}
-	if holder, err := store.TicketHolder(t.Context(), tokenHash("k1")); err != nil || holder != bot {
+	if holder, err := store.TicketHolder(t.Context(), tokenHash("k1")); err != nil || !reflect.DeepEqual(holder, bot) {
 		t.Errorf("设置密码之后 Key 的持有者 = %#v, %v，期望 Key 仍然有效", holder, err)
 	}
 	if n := ticketCount(t, pool, id); n != 1 {

internal/adapter/out/postgres/internal/query/members.sql.go

@@ -62,8 +62,9 @@ func (q *Queries) CreateSession(ctx context.Context, arg CreateSessionParams) er
 	return err
 }
 
-const deleteMemberSessions = `-- name: DeleteMemberSessions :exec
+const deleteMemberSessions = `-- name: DeleteMemberSessions :many
 DELETE FROM member_sessions WHERE member_id = $1 AND token_hash IS DISTINCT FROM $2::bytea
+RETURNING token_hash
 `
 
 type DeleteMemberSessionsParams struct {
@@ -72,9 +73,24 @@ type DeleteMemberSessionsParams struct {
 }
 
 // keep 为 NULL 时删除这个成员的全部会话。
-func (q *Queries) DeleteMemberSessions(ctx context.Context, arg DeleteMemberSessionsParams) error {
-	_, err := q.db.Exec(ctx, deleteMemberSessions, arg.MemberID, arg.Keep)
-	return err
+func (q *Queries) DeleteMemberSessions(ctx context.Context, arg DeleteMemberSessionsParams) ([][]byte, error) {
+	rows, err := q.db.Query(ctx, deleteMemberSessions, arg.MemberID, arg.Keep)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items [][]byte
+	for rows.Next() {
+		var token_hash []byte
+		if err := rows.Scan(&token_hash); err != nil {
+			return nil, err
+		}
+		items = append(items, token_hash)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
 }
 
 const deleteSession = `-- name: DeleteSession :execrows

internal/adapter/out/postgres/internal/query/tickets.sql.go

@@ -10,17 +10,25 @@ import (
 )
 
 const createTicket = `-- name: CreateTicket :exec
-INSERT INTO room_tickets (token_hash, room_id, member_id) VALUES ($1, $2, $3)
+INSERT INTO room_tickets (token_hash, room_id, member_id, session_token_hash)
+VALUES ($1, $2, $3, $4)
 `
 
 type CreateTicketParams struct {
-	TokenHash []byte
-	RoomID    string
-	MemberID  string
+	TokenHash        []byte
+	RoomID           string
+	MemberID         string
+	SessionTokenHash []byte
 }
 
+// bot 的 API Key 没有会话，session_token_hash 为 NULL。
 func (q *Queries) CreateTicket(ctx context.Context, arg CreateTicketParams) error {
-	_, err := q.db.Exec(ctx, createTicket, arg.TokenHash, arg.RoomID, arg.MemberID)
+	_, err := q.db.Exec(ctx, createTicket,
+		arg.TokenHash,
+		arg.RoomID,
+		arg.MemberID,
+		arg.SessionTokenHash,
+	)
 	return err
 }
 
@@ -37,7 +45,7 @@ func (q *Queries) LockRoomPassword(ctx context.Context, id string) (string, erro
 }
 
 const ticketHolder = `-- name: TicketHolder :one
-SELECT t.room_id, m.id, m.name, m.avatar_hash, r.bot_member_id IS NOT DISTINCT FROM m.id AS bot
+SELECT t.room_id, m.id, m.name, m.avatar_hash, r.bot_member_id IS NOT DISTINCT FROM m.id AS bot, t.session_token_hash
 FROM room_tickets t
 JOIN members m ON m.id = t.member_id
 JOIN rooms r ON r.id = t.room_id
@@ -45,11 +53,12 @@ WHERE t.token_hash = $1
 `
 
 type TicketHolderRow struct {
-	RoomID     string
-	ID         string
-	Name       string
-	AvatarHash *string
-	Bot        bool
+	RoomID           string
+	ID               string
+	Name             string
+	AvatarHash       *string
+	Bot              bool
+	SessionTokenHash []byte
 }
 
 func (q *Queries) TicketHolder(ctx context.Context, tokenHash []byte) (TicketHolderRow, error) {
@@ -61,6 +70,7 @@ func (q *Queries) TicketHolder(ctx context.Context, tokenHash []byte) (TicketHol
 		&i.Name,
 		&i.AvatarHash,
 		&i.Bot,
+		&i.SessionTokenHash,
 	)
 	return i, err
 }

internal/adapter/out/postgres/members.go

@@ -131,7 +131,8 @@ func (s *Store) DeleteSession(ctx context.Context, tokenHash []byte) error {
 	return nil
 }
 
-func (s *Store) SetPassword(ctx context.Context, memberID, passwordHash string, keepTokenHash []byte) error {
+func (s *Store) SetPassword(ctx context.Context, memberID, passwordHash string, keepTokenHash []byte) ([][]byte, error) {
+	var deleted [][]byte
 	err := pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
 		q := s.q.WithTx(tx)
 		n, err := q.SetMemberPassword(ctx, query.SetMemberPasswordParams{ID: memberID, PasswordHash: &passwordHash})
@@ -141,10 +142,11 @@ func (s *Store) SetPassword(ctx context.Context, memberID, passwordHash string,
 		if n == 0 {
 			return port.ErrMemberNotFound
 		}
-		return q.DeleteMemberSessions(ctx, query.DeleteMemberSessionsParams{MemberID: memberID, Keep: keepTokenHash})
+		deleted, err = q.DeleteMemberSessions(ctx, query.DeleteMemberSessionsParams{MemberID: memberID, Keep: keepTokenHash})
+		return err
 	})
 	if err != nil {
-		return fmt.Errorf("修改成员 %s 的密码：%w", memberID, err)
+		return nil, fmt.Errorf("修改成员 %s 的密码：%w", memberID, err)
 	}
-	return nil
+	return deleted, nil
 }

internal/adapter/out/postgres/members_test.go

@@ -226,8 +226,9 @@ func TestMemberSessions(t *testing.T) {
 	if err := store.Register(t.Context(), alice.ID, "alice", "old"); err != nil {
 		t.Fatal(err)
 	}
-	if err := store.SetPassword(t.Context(), alice.ID, "new", tokenHash("alice-1")); err != nil {
-		t.Fatal(err)
+	deleted, err := store.SetPassword(t.Context(), alice.ID, "new", tokenHash("alice-1"))
+	if want := [][]byte{tokenHash("alice-2"), tokenHash("alice-3")}; err != nil || !sameHashes(deleted, want) {
+		t.Errorf("SetPassword 返回 %x, %v，期望被删除的 alice-2、alice-3", deleted, err)
 	}
 	if ownerOf(t, store, "alice-1") != alice.ID || ownerOf(t, store, "alice-2") != "" || ownerOf(t, store, "alice-3") != "" {
 		t.Error("SetPassword 应只留下指定的会话")
@@ -243,7 +244,7 @@ func TestMemberSessions(t *testing.T) {
 func TestSetPassword(t *testing.T) {
 	store, _ := newStore(t)
 	guest := createMember(t, store, "guest", "游客-1a2b")
-	if err := store.SetPassword(t.Context(), guest.ID, "hash", nil); !errors.Is(err, port.ErrMemberNotFound) {
+	if _, err := store.SetPassword(t.Context(), guest.ID, "hash", nil); !errors.Is(err, port.ErrMemberNotFound) {
 		t.Errorf("给游客设置密码 = %v，期望 ErrMemberNotFound", err)
 	}
 	alice := createMember(t, store, "alice-1", "游客-3c4d")
@@ -254,8 +255,8 @@ func TestSetPassword(t *testing.T) {
 	if err := store.ReplaceSession(t.Context(), tokenHash("other"), alice.ID, tokenHash("alice-2"), "old"); err != nil {
 		t.Fatal(err)
 	}
-	if err := store.SetPassword(t.Context(), alice.ID, "new", tokenHash("alice-1")); err != nil {
-		t.Fatal(err)
+	if deleted, err := store.SetPassword(t.Context(), alice.ID, "new", tokenHash("alice-1")); err != nil || !sameHashes(deleted, [][]byte{tokenHash("alice-2")}) {
+		t.Errorf("SetPassword 返回 %x, %v，期望被删除的 alice-2", deleted, err)
 	}
 	if account, err := store.AccountByUsername(t.Context(), "alice"); err != nil || account.PasswordHash != "new" {
 		t.Errorf("修改之后的密码哈希 = %q, %v，期望 new", account.PasswordHash, err)
@@ -263,8 +264,8 @@ func TestSetPassword(t *testing.T) {
 	if ownerOf(t, store, "alice-1") != alice.ID || ownerOf(t, store, "alice-2") != "" {
 		t.Error("修改密码应只留下指定的会话")
 	}
-	if err := store.SetPassword(t.Context(), alice.ID, "newer", nil); err != nil {
-		t.Fatal(err)
+	if deleted, err := store.SetPassword(t.Context(), alice.ID, "newer", nil); err != nil || !sameHashes(deleted, [][]byte{tokenHash("alice-1")}) {
+		t.Errorf("SetPassword 返回 %x, %v，期望被删除的 alice-1", deleted, err)
 	}
 	if got := ownerOf(t, store, "alice-1"); got != "" {
 		t.Errorf("不指定保留的会话时应删除全部会话，alice-1 仍属于 %s", got)

internal/adapter/out/postgres/migrate_test.go

@@ -179,3 +179,49 @@ func TestMemberSessionsMigrationKeepsMemberTokens(t *testing.T) {
 		t.Errorf("bot 成员原来的 token 在迁移之后 = %v，期望不再有效", err)
 	}
 }
+
+func TestTicketSessionsMigrationBindsTicketsToSessions(t *testing.T) {
+	ctx := t.Context()
+	name := fmt.Sprintf("test_%d", databases.Add(1))
+	if _, err := admin.Exec(ctx, "CREATE DATABASE "+name); err != nil {
+		t.Fatal(err)
+	}
+	cfg := server.Copy()
+	cfg.ConnConfig.Database = name
+	pool := connect(t, cfg)
+	db := stdlib.OpenDBFromPool(pool)
+	t.Cleanup(func() { _ = db.Close() })
+	provider, err := goose.NewProvider(goose.DialectPostgres, db, migrations.FS)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := provider.UpTo(ctx, 2); err != nil {
+		t.Fatal(err)
+	}
+	var guest, bot, roomID string
+	if err := pool.QueryRow(ctx, "INSERT INTO members (token_hash, name) VALUES ($1, '游客-1a2b') RETURNING id", tokenHash("guest")).Scan(&guest); err != nil {
+		t.Fatal(err)
+	}
+	if err := pool.QueryRow(ctx, "INSERT INTO members (token_hash, name) VALUES ($1, 'bot') RETURNING id", tokenHash("bot")).Scan(&bot); err != nil {
+		t.Fatal(err)
+	}
+	if err := pool.QueryRow(ctx, "INSERT INTO rooms (kind, name, description, bot_member_id) VALUES ('persistent', '大厅', '', $1) RETURNING id", bot).Scan(&roomID); err != nil {
+		t.Fatal(err)
+	}
+	for ticket, memberID := range map[string]string{"guest-ticket": guest, "api-key": bot} {
+		if _, err := pool.Exec(ctx, "INSERT INTO room_tickets (token_hash, room_id, member_id) VALUES ($1, $2, $3)", tokenHash(ticket), roomID, memberID); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := provider.Up(ctx); err != nil {
+		t.Fatal(err)
+	}
+
+	store := New(pool)
+	if holder, err := store.TicketHolder(ctx, tokenHash("guest-ticket")); err != nil || !bytes.Equal(holder.SessionTokenHash, tokenHash("guest")) {
+		t.Errorf("游客的票据 = %#v, %v，期望归到游客的会话", holder, err)
+	}
+	if holder, err := store.TicketHolder(ctx, tokenHash("api-key")); err != nil || !holder.Bot || holder.SessionTokenHash != nil {
+		t.Errorf("API Key 的票据 = %#v, %v，期望保留，没有会话", holder, err)
+	}
+}

internal/adapter/out/postgres/tickets.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 
 	"github.com/dihemoe/alisten/internal/adapter/out/postgres/internal/query"
 	"github.com/dihemoe/alisten/internal/app/port"
@@ -27,10 +28,10 @@ func (s *Store) CreateTicket(ctx context.Context, t port.NewTicket) error {
 		if hash != t.PasswordHash {
 			return port.ErrPasswordChanged
 		}
-		return q.CreateTicket(ctx, query.CreateTicketParams{TokenHash: t.TokenHash, RoomID: t.RoomID, MemberID: t.MemberID})
+		return q.CreateTicket(ctx, query.CreateTicketParams{TokenHash: t.TokenHash, RoomID: t.RoomID, MemberID: t.MemberID, SessionTokenHash: t.SessionTokenHash})
 	})
 	if err != nil {
-		return fmt.Errorf("签发房间 %s 的票据：%w", t.RoomID, err)
+		return fmt.Errorf("签发房间 %s 的票据：%w", t.RoomID, sessionError(err))
 	}
 	return nil
 }
@@ -44,8 +45,19 @@ func (s *Store) TicketHolder(ctx context.Context, tokenHash []byte) (port.Ticket
 		return port.TicketHolder{}, fmt.Errorf("查找票据：%w", err)
 	}
 	return port.TicketHolder{
-		RoomID: row.RoomID,
-		Member: member.Member{ID: row.ID, Name: row.Name, AvatarHash: orEmpty(row.AvatarHash)},
-		Bot:    row.Bot,
+		RoomID:           row.RoomID,
+		Member:           member.Member{ID: row.ID, Name: row.Name, AvatarHash: orEmpty(row.AvatarHash)},
+		Bot:              row.Bot,
+		SessionTokenHash: row.SessionTokenHash,
 	}, nil
 }
+
+const foreignKeyViolation = "23503"
+
+func sessionError(err error) error {
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) && pgErr.Code == foreignKeyViolation && pgErr.ConstraintName == "room_tickets_session_token_hash_fkey" {
+		return port.ErrSessionNotFound
+	}
+	return err
+}

internal/adapter/out/postgres/tickets_test.go

@@ -2,6 +2,8 @@ package postgres
 
 import (
 	"errors"
+	"reflect"
+	"slices"
 	"testing"
 	"time"
 
@@ -16,11 +18,11 @@ func TestTicketHolder(t *testing.T) {
 	alice := createMember(t, store, "alice", "甲")
 	id := createRoom(t, store, port.NewRoom{Kind: room.Ephemeral, Name: "临时"})
 
-	if err := store.CreateTicket(t.Context(), port.NewTicket{TokenHash: tokenHash("t1"), RoomID: id, MemberID: alice.ID}); err != nil {
+	if err := store.CreateTicket(t.Context(), port.NewTicket{TokenHash: tokenHash("t1"), RoomID: id, MemberID: alice.ID, SessionTokenHash: tokenHash("alice")}); err != nil {
 		t.Fatal(err)
 	}
 	got, err := store.TicketHolder(t.Context(), tokenHash("t1"))
-	if want := (port.TicketHolder{RoomID: id, Member: alice}); err != nil || got != want {
+	if want := (port.TicketHolder{RoomID: id, Member: alice, SessionTokenHash: tokenHash("alice")}); err != nil || !reflect.DeepEqual(got, want) {
 		t.Errorf("TicketHolder = %#v, %v，期望 %#v", got, err, want)
 	}
 
@@ -37,6 +39,87 @@ func TestTicketHolder(t *testing.T) {
 	}
 }
 
+func TestCreateTicketRejectsDeletedSession(t *testing.T) {
+	store, _ := newStore(t)
+	alice := createMember(t, store, "alice", "甲")
+	id := createRoom(t, store, port.NewRoom{Kind: room.Ephemeral, Name: "临时"})
+	if err := store.DeleteSession(t.Context(), tokenHash("alice")); err != nil {
+		t.Fatal(err)
+	}
+
+	err := store.CreateTicket(t.Context(), port.NewTicket{TokenHash: tokenHash("t1"), RoomID: id, MemberID: alice.ID, SessionTokenHash: tokenHash("alice")})
+	if !errors.Is(err, port.ErrSessionNotFound) {
+		t.Errorf("会话已经删除时 CreateTicket = %v，期望 ErrSessionNotFound", err)
+	}
+}
+
+func TestTicketsBelongToSessions(t *testing.T) {
+	store, _ := newStore(t)
+	alice := createMember(t, store, "alice-1", "甲")
+	id := createRoom(t, store, port.NewRoom{Kind: room.Ephemeral, Name: "临时"})
+	for _, guest := range []string{"guest-2", "guest-3", "guest-4"} {
+		createMember(t, store, guest, "游客")
+	}
+	for _, s := range []struct{ guest, token string }{{"guest-2", "alice-2"}, {"guest-3", "alice-3"}} {
+		if err := store.ReplaceSession(t.Context(), tokenHash(s.guest), alice.ID, tokenHash(s.token), ""); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for _, session := range []string{"alice-1", "alice-2", "alice-3", "guest-4"} {
+		memberID := alice.ID
+		if session == "guest-4" {
+			memberID = ownerOf(t, store, "guest-4")
+		}
+		if err := store.CreateTicket(t.Context(), port.NewTicket{TokenHash: tokenHash("ticket-" + session), RoomID: id, MemberID: memberID, SessionTokenHash: tokenHash(session)}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	valid := func(session string) bool {
+		t.Helper()
+		_, err := store.TicketHolder(t.Context(), tokenHash("ticket-"+session))
+		if err != nil && !errors.Is(err, port.ErrTicketNotFound) {
+			t.Fatal(err)
+		}
+		return err == nil
+	}
+
+	if err := store.Register(t.Context(), alice.ID, "alice", "old"); err != nil {
+		t.Fatal(err)
+	}
+	deleted, err := store.SetPassword(t.Context(), alice.ID, "new", tokenHash("alice-1"))
+	if want := [][]byte{tokenHash("alice-2"), tokenHash("alice-3")}; err != nil || !sameHashes(deleted, want) {
+		t.Errorf("SetPassword 返回 %x, %v，期望被删除的 alice-2、alice-3", deleted, err)
+	}
+	if !valid("alice-1") || valid("alice-2") || valid("alice-3") {
+		t.Error("删除会话之后，只有保留的会话的票据仍然有效")
+	}
+	if err := store.ReplaceSession(t.Context(), tokenHash("guest-4"), alice.ID, tokenHash("alice-4"), "new"); err != nil {
+		t.Fatal(err)
+	}
+	if valid("guest-4") {
+		t.Error("登录换掉游客的会话之后，游客的票据仍然有效")
+	}
+	if err := store.DeleteSession(t.Context(), tokenHash("alice-1")); err != nil {
+		t.Fatal(err)
+	}
+	if valid("alice-1") {
+		t.Error("退出登录之后，这个会话的票据仍然有效")
+	}
+}
+
+// 不计顺序。
+func sameHashes(got, want [][]byte) bool {
+	key := func(hashes [][]byte) []string {
+		s := make([]string, len(hashes))
+		for i, h := range hashes {
+			s[i] = string(h)
+		}
+		slices.Sort(s)
+		return s
+	}
+	return slices.Equal(key(got), key(want))
+}
+
 func TestCreateTicketChecksPasswordHash(t *testing.T) {
 	store, _ := newStore(t)
 	alice := createMember(t, store, "alice", "甲")

internal/adapter/out/redis/bus.go

@@ -198,6 +198,15 @@ func (b *Bus) PublishMember(ctx context.Context, m member.Member) error {
 	return err
 }
 
+func (b *Bus) PublishSessionsRevoked(ctx context.Context, sessionTokenHashes [][]byte) error {
+	data, err := json.Marshal(sessionTokenHashes)
+	if err != nil {
+		return err
+	}
+	_, err = b.publish(ctx, membersChannel, kindSessionsRevoked, data, nil)
+	return err
+}
+
 func (b *Bus) PublishPrepared(ctx context.Context, o port.PrepareOutcome) error {
 	data, err := json.Marshal(o)
 	if err != nil {
@@ -362,6 +371,10 @@ func (b *Bus) delivery(channel string, e envelope) (port.Delivery, error) {
 		var m member.Member
 		err := json.Unmarshal(e.Data, &m)
 		return port.MemberDelivery{Member: m}, err
+	case channel == membersChannel && e.Kind == kindSessionsRevoked:
+		var hashes [][]byte
+		err := json.Unmarshal(e.Data, &hashes)
+		return port.SessionsRevokedDelivery{SessionTokenHashes: hashes}, err
 	case channel == preparedChannel && e.Kind == kindPrepared:
 		var o port.PrepareOutcome
 		err := json.Unmarshal(e.Data, &o)

internal/adapter/out/redis/bus_test.go

@@ -344,6 +344,25 @@ func TestPreparedOutcomes(t *testing.T) {
 	}
 }
 
+func TestSessionsRevoked(t *testing.T) {
+	client := newClient(t)
+	a, b := newBus(t, client, "v1"), newBus(t, client, "v1")
+	for _, f := range []*busFixture{a, b} {
+		if err := f.bus.SubscribeInstance(t.Context()); err != nil {
+			t.Fatal(err)
+		}
+	}
+	hashes := [][]byte{{0x01, 0x02}, {0xff, 0x00, 0x7f}}
+	if err := a.bus.PublishSessionsRevoked(t.Context(), hashes); err != nil {
+		t.Fatal(err)
+	}
+	for _, f := range []*busFixture{a, b} {
+		if got := f.next(t); !reflect.DeepEqual(got, port.SessionsRevokedDelivery{SessionTokenHashes: hashes}) {
+			t.Errorf("收到 %#v，期望被删除的会话 %x", got, hashes)
+		}
+	}
+}
+
 func TestUnsubscribeRoom(t *testing.T) {
 	client := newClient(t)
 	a, b := newBus(t, client, "v1"), newBus(t, client, "v1")

internal/adapter/out/redis/codec.go

@@ -55,11 +55,12 @@ type (
 )
 
 const (
-	kindEmitted  = "Emitted"
-	kindReplied  = "Replied"
-	kindInput    = "RoomInput"
-	kindMember   = "Member"
-	kindPrepared = "Prepared"
+	kindEmitted         = "Emitted"
+	kindReplied         = "Replied"
+	kindInput           = "RoomInput"
+	kindMember          = "Member"
+	kindSessionsRevoked = "SessionsRevoked"
+	kindPrepared        = "Prepared"
 	// room.Command 的 Action 是接口，不能像 inputs 中的类型那样直接编码。
 	inputCommand = "Command"
 )

internal/app/admin/admin_test.go

@@ -5,6 +5,7 @@ import (
 	"crypto/sha256"
 	"errors"
 	"log/slog"
+	"reflect"
 	"strconv"
 	"strings"
 	"testing"
@@ -149,7 +150,7 @@ func (f fixture) enter(t *testing.T, roomID, name string) ([]byte, *realtime.Ses
 	if err := f.store.CreateTicket(t.Context(), port.NewTicket{TokenHash: hash[:], RoomID: roomID, MemberID: m.ID, PasswordHash: r.PasswordHash}); err != nil {
 		t.Fatal(err)
 	}
-	session, err := f.gateway.Join(t.Context(), roomID, m)
+	session, err := f.gateway.Join(t.Context(), roomID, m, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -322,7 +323,7 @@ func TestRoomAPIKey(t *testing.T) {
 	if _, err := holder(key); !errors.Is(err, port.ErrTicketNotFound) {
 		t.Errorf("重置之后旧 Key 的错误 = %v，期望 ErrTicketNotFound", err)
 	}
-	if got, err := holder(reset); err != nil || got != bot {
+	if got, err := holder(reset); err != nil || !reflect.DeepEqual(got, bot) {
 		t.Errorf("重置之后新 Key 的持有者 = %+v, %v，期望 bot 成员不变", got, err)
 	}
 
@@ -331,7 +332,7 @@ func TestRoomAPIKey(t *testing.T) {
 		if err := f.admin.UpdateRoom(t.Context(), id, RoomChange{Password: ptr(password)}); err != nil {
 			t.Fatal(err)
 		}
-		if got, err := holder(reset); err != nil || got != bot {
+		if got, err := holder(reset); err != nil || !reflect.DeepEqual(got, bot) {
 			t.Errorf("密码改为 %q 之后 Key 的持有者 = %+v, %v，期望 Key 仍然有效", password, got, err)
 		}
 	}

internal/app/identity/account.go

@@ -3,9 +3,11 @@ package identity
 import (
 	"context"
 	"errors"
+	"log/slog"
 	"time"
 
 	"github.com/dihemoe/alisten/internal/app/port"
+	"github.com/dihemoe/alisten/internal/app/realtime"
 	"github.com/dihemoe/alisten/internal/domain/member"
 )
 
@@ -36,14 +38,18 @@ type Account struct {
 	Email    string
 }
 
+// 删除会话之后，用这些会话的票据建立的房间连接立即断开；数据已经修改而断开的通知没有发出时返回错误，
+// 只有撤销发起请求的这台设备自己的会话（Login、Logout）例外，只记日志。
 type Accounts struct {
 	members port.Members
 	site    port.Site
 	quota   port.AccountQuota
+	gateway *realtime.Gateway
+	logger  *slog.Logger
 }
 
-func NewAccounts(members port.Members, site port.Site, quota port.AccountQuota) *Accounts {
-	return &Accounts{members: members, site: site, quota: quota}
+func NewAccounts(members port.Members, site port.Site, quota port.AccountQuota, gateway *realtime.Gateway, logger *slog.Logger) *Accounts {
+	return &Accounts{members: members, site: site, quota: quota, gateway: gateway, logger: logger}
 }
 
 // 游客没有账号，ok 为 false。
@@ -96,7 +102,8 @@ func (a *Accounts) Login(ctx context.Context, current member.Member, currentToke
 		return Session{}, Account{}, err
 	}
 	token, hash := newSecret()
-	err = a.members.ReplaceSession(ctx, hashSecret(currentToken), account.Member.ID, hash, account.PasswordHash)
+	guest := hashSecret(currentToken)
+	err = a.members.ReplaceSession(ctx, guest, account.Member.ID, hash, account.PasswordHash)
 	switch {
 	case errors.Is(err, port.ErrSessionNotFound):
 		return Session{}, Account{}, ErrInvalidMemberToken
@@ -105,6 +112,10 @@ func (a *Accounts) Login(ctx context.Context, current member.Member, currentToke
 	case err != nil:
 		return Session{}, Account{}, err
 	}
+	// 页面随后会重新加载，游客的连接即使没断开也不再使用；此时返回错误会让用户拿不到已经签发的新会话。
+	if err := a.gateway.RevokeSessions(ctx, [][]byte{guest}); err != nil {
+		a.logger.WarnContext(ctx, "登录后断开游客会话的通知没有发出", "error", err)
+	}
 	return Session{MemberToken: token, Member: account.Member}, Account{Username: account.Username, Email: account.Email}, nil
 }
 
@@ -134,7 +145,15 @@ func (a *Accounts) verify(ctx context.Context, username, password string) (port.
 }
 
 func (a *Accounts) Logout(ctx context.Context, token string) error {
-	return a.members.DeleteSession(ctx, hashSecret(token))
+	hash := hashSecret(token)
+	if err := a.members.DeleteSession(ctx, hash); err != nil {
+		return err
+	}
+	// 页面随后会重新加载，不因通知没有发出而让退出失败。
+	if err := a.gateway.RevokeSessions(ctx, [][]byte{hash}); err != nil {
+		a.logger.WarnContext(ctx, "退出后断开会话的通知没有发出", "error", err)
+	}
+	return nil
 }
 
 // 修改成功后，账号除了发出请求的这个会话，其他会话都失效。
@@ -159,7 +178,11 @@ func (a *Accounts) ChangePassword(ctx context.Context, memberID, token, currentP
 	if err != nil {
 		return err
 	}
-	return a.members.SetPassword(ctx, memberID, hash, hashSecret(token))
+	deleted, err := a.members.SetPassword(ctx, memberID, hash, hashSecret(token))
+	if err != nil {
+		return err
+	}
+	return a.gateway.RevokeSessions(ctx, deleted)
 }
 
 // Find 和 ResetPassword 只供后台使用，调用方要先校验管理 token。
@@ -182,7 +205,11 @@ func (a *Accounts) ResetPassword(ctx context.Context, username string) (string,
 	if err != nil {
 		return "", err
 	}
-	if err := a.members.SetPassword(ctx, account.Member.ID, hash, nil); err != nil {
+	deleted, err := a.members.SetPassword(ctx, account.Member.ID, hash, nil)
+	if err != nil {
+		return "", err
+	}
+	if err := a.gateway.RevokeSessions(ctx, deleted); err != nil {
 		return "", err
 	}
 	return password, nil

internal/app/identity/account_test.go

@@ -1,9 +1,11 @@
 package identity
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
+	"log/slog"
 	"regexp"
 	"slices"
 	"sync"
@@ -12,6 +14,7 @@ import (
 
 	"github.com/dihemoe/alisten/internal/app/port"
 	"github.com/dihemoe/alisten/internal/app/port/porttest"
+	"github.com/dihemoe/alisten/internal/app/realtime"
 	"github.com/dihemoe/alisten/internal/domain/member"
 )
 
@@ -36,11 +39,65 @@ func (q *fakeQuota) Take(_ context.Context, key string, burst int, refill time.D
 	return 0, nil
 }
 
-func newAccounts(t *testing.T) (*Accounts, *Service, *porttest.Store, *fakeQuota) {
+// revoked 返回此前广播过的被删除的会话。
+func newAccounts(t *testing.T) (a *Accounts, s *Service, store *porttest.Store, q *fakeQuota, revoked func() [][]byte) {
 	t.Helper()
-	s, store := newService(t)
-	q := &fakeQuota{exhausted: map[string]bool{}}
-	return NewAccounts(store, store, q), s, store, q
+	store = porttest.New()
+	node := porttest.NewRedis().Node("a")
+	if err := node.SubscribeInstance(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	gateway := newGateway(store, node, node)
+	q = &fakeQuota{exhausted: map[string]bool{}}
+	revoked = func() [][]byte {
+		t.Helper()
+		done, cancel := context.WithCancel(t.Context())
+		cancel()
+		var hashes [][]byte
+		for {
+			d, err := node.Receive(done)
+			if err != nil {
+				return hashes
+			}
+			if r, ok := d.(port.SessionsRevokedDelivery); ok {
+				hashes = append(hashes, r.SessionTokenHashes...)
+			}
+		}
+	}
+	return NewAccounts(store, store, q, gateway, slog.New(slog.DiscardHandler)), New(store, store, store, gateway), store, q, revoked
+}
+
+// 不计顺序。
+func sameHashes(got [][]byte, tokens ...string) bool {
+	want := make([]string, len(tokens))
+	for i, token := range tokens {
+		want[i] = string(hashSecret(token))
+	}
+	keys := make([]string, len(got))
+	for i, h := range got {
+		keys[i] = string(h)
+	}
+	slices.Sort(want)
+	slices.Sort(keys)
+	return slices.Equal(keys, want)
+}
+
+func issueTicket(t *testing.T, s *Service, session Session, roomID string) string {
+	t.Helper()
+	ticket, err := s.IssueTicket(t.Context(), session.Member.ID, session.MemberToken, roomID, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return ticket
+}
+
+func ticketValid(t *testing.T, s *Service, ticket string) bool {
+	t.Helper()
+	_, err := s.VerifyTicket(t.Context(), ticket)
+	if err != nil && !errors.Is(err, ErrInvalidTicket) {
+		t.Fatal(err)
+	}
+	return err == nil
 }
 
 // 返回 token 当前所属的成员；token 已经失效时 ok 为 false。
@@ -66,7 +123,7 @@ func registerAccount(t *testing.T, a *Accounts, s *Service, username, password s
 }
 
 func TestRegister(t *testing.T) {
-	a, s, store, q := newAccounts(t)
+	a, s, store, q, _ := newAccounts(t)
 	guest := createMember(t, s)
 	account, err := a.Register(t.Context(), guest.Member.ID, clientIP, "alice", "correct horse")
 	if err != nil || account != (Account{Username: "alice"}) {
@@ -91,7 +148,7 @@ func TestRegister(t *testing.T) {
 }
 
 func TestAccountKeepsEmail(t *testing.T) {
-	a, s, _, _ := newAccounts(t)
+	a, s, _, _, _ := newAccounts(t)
 	alice := registerAccount(t, a, s, "alice", "correct horse")
 	guest := createMember(t, s)
 
@@ -122,14 +179,14 @@ func TestAccountKeepsEmail(t *testing.T) {
 }
 
 func TestGuestHasNoAccount(t *testing.T) {
-	a, s, _, _ := newAccounts(t)
+	a, s, _, _, _ := newAccounts(t)
 	if _, ok, err := a.Account(t.Context(), createMember(t, s).Member.ID); err != nil || ok {
 		t.Errorf("游客的 Account = %v, %v，期望没有账号", ok, err)
 	}
 }
 
 func TestRegisterRejected(t *testing.T) {
-	a, s, store, q := newAccounts(t)
+	a, s, store, q, _ := newAccounts(t)
 	alice := registerAccount(t, a, s, "alice", "correct horse")
 	if _, err := a.Register(t.Context(), createMember(t, s).Member.ID, clientIP, "alice", "another horse"); !errors.Is(err, port.ErrUsernameTaken) {
 		t.Errorf("用户名已被占用时 = %v，期望 ErrUsernameTaken", err)
@@ -160,14 +217,22 @@ func TestRegisterRejected(t *testing.T) {
 }
 
 func TestLogin(t *testing.T) {
-	a, s, store, q := newAccounts(t)
+	a, s, store, q, revoked := newAccounts(t)
 	alice := registerAccount(t, a, s, "alice", "correct horse")
 	guest := createMember(t, s)
+	id := createRoom(t, store, "")
+	guestTicket, aliceTicket := issueTicket(t, s, guest, id), issueTicket(t, s, alice, id)
 	q.taken = nil
 	session, account, err := a.Login(t.Context(), guest.Member, guest.MemberToken, clientIP, "alice", "correct horse")
 	if err != nil {
 		t.Fatal(err)
 	}
+	if got := revoked(); !sameHashes(got, guest.MemberToken) {
+		t.Errorf("广播的被删除的会话 = %x，期望游客原来的会话", got)
+	}
+	if ticketValid(t, s, guestTicket) || !ticketValid(t, s, aliceTicket) {
+		t.Error("登录之后只有游客原来的会话的票据失效")
+	}
 	if session.Member != alice.Member || account != (Account{Username: "alice"}) {
 		t.Errorf("Login = %#v, %#v，期望登录 alice", session.Member, account)
 	}
@@ -189,7 +254,7 @@ func TestLogin(t *testing.T) {
 }
 
 func TestLoginRejected(t *testing.T) {
-	a, s, store, q := newAccounts(t)
+	a, s, store, q, _ := newAccounts(t)
 	alice := registerAccount(t, a, s, "alice", "correct horse")
 	guest := createMember(t, s)
 	for _, tc := range []struct{ name, username, password string }{
@@ -217,24 +282,36 @@ func TestLoginRejected(t *testing.T) {
 }
 
 func TestLogout(t *testing.T) {
-	a, s, store, _ := newAccounts(t)
+	a, s, store, _, revoked := newAccounts(t)
 	alice := registerAccount(t, a, s, "alice", "correct horse")
+	id := createRoom(t, store, "")
+	aliceTicket := issueTicket(t, s, alice, id)
+	revoked()
 	if err := a.Logout(t.Context(), alice.MemberToken); err != nil {
 		t.Fatal(err)
 	}
 	if _, ok := tokenOwner(t, store, alice.MemberToken); ok {
 		t.Error("退出登录之后 token 应失效")
 	}
+	if got := revoked(); !sameHashes(got, alice.MemberToken) {
+		t.Errorf("退出登录之后广播的会话 = %x，期望当前会话", got)
+	}
+	if ticketValid(t, s, aliceTicket) {
+		t.Error("退出登录之后这个会话的票据应失效")
+	}
 }
 
 func TestChangePassword(t *testing.T) {
-	a, s, store, q := newAccounts(t)
+	a, s, store, q, revoked := newAccounts(t)
 	alice := registerAccount(t, a, s, "alice", "correct horse")
 	guest := createMember(t, s)
 	other, _, err := a.Login(t.Context(), guest.Member, guest.MemberToken, clientIP, "alice", "correct horse")
 	if err != nil {
 		t.Fatal(err)
 	}
+	id := createRoom(t, store, "")
+	aliceTicket, otherTicket := issueTicket(t, s, alice, id), issueTicket(t, s, other, id)
+	revoked()
 	q.taken = nil
 	if err := a.ChangePassword(t.Context(), alice.Member.ID, alice.MemberToken, "wrong horse", "battery staple"); !errors.Is(err, ErrWrongCurrentPassword) {
 		t.Errorf("当前密码错误时 = %v，期望 ErrWrongCurrentPassword", err)
@@ -251,6 +328,12 @@ func TestChangePassword(t *testing.T) {
 	if _, ok := tokenOwner(t, store, other.MemberToken); ok {
 		t.Error("修改密码之后其他会话应失效")
 	}
+	if got := revoked(); !sameHashes(got, other.MemberToken) {
+		t.Errorf("修改密码之后广播的会话 = %x，期望其他设备的会话", got)
+	}
+	if !ticketValid(t, s, aliceTicket) || ticketValid(t, s, otherTicket) {
+		t.Error("修改密码之后只有其他设备的票据失效")
+	}
 	next := createMember(t, s)
 	if _, _, err := a.Login(t.Context(), next.Member, next.MemberToken, clientIP, "alice", "correct horse"); !errors.Is(err, ErrInvalidCredentials) {
 		t.Errorf("用旧密码登录 = %v，期望 ErrInvalidCredentials", err)
@@ -265,8 +348,9 @@ func TestChangePassword(t *testing.T) {
 }
 
 func TestResetPassword(t *testing.T) {
-	a, s, store, _ := newAccounts(t)
+	a, s, store, _, revoked := newAccounts(t)
 	alice := registerAccount(t, a, s, "alice", "correct horse")
+	aliceTicket := issueTicket(t, s, alice, createRoom(t, store, ""))
 	if _, err := a.Find(t.Context(), "nobody"); !errors.Is(err, ErrAccountNotFound) {
 		t.Errorf("Find 不存在的用户名 = %v，期望 ErrAccountNotFound", err)
 	}
@@ -286,26 +370,107 @@ func TestResetPassword(t *testing.T) {
 	if _, ok := tokenOwner(t, store, alice.MemberToken); ok {
 		t.Error("重置密码之后账号的所有会话应失效")
 	}
+	if got := revoked(); !sameHashes(got, alice.MemberToken) {
+		t.Errorf("重置密码之后广播的会话 = %x，期望账号的所有会话", got)
+	}
+	if ticketValid(t, s, aliceTicket) {
+		t.Error("重置密码之后账号的票据应失效")
+	}
 	guest := createMember(t, s)
 	if _, _, err := a.Login(t.Context(), guest.Member, guest.MemberToken, clientIP, "alice", password); err != nil {
 		t.Errorf("用重置后的密码登录 = %v", err)
 	}
 }
 
+type noRevocation struct {
+	port.Bus
+}
+
+func (noRevocation) PublishSessionsRevoked(context.Context, [][]byte) error {
+	return porttest.ErrRedisDown
+}
+
+// 撤销的是别的设备的会话：断开的通知没有发出时报错，让用户知道它们没有被踢出去。
+func TestOtherDeviceOperationsFailWhenRevocationFails(t *testing.T) {
+	store := porttest.New()
+	node := porttest.NewRedis().Node("a")
+	gateway := newGateway(store, node, noRevocation{node})
+	s := New(store, store, store, gateway)
+	a := NewAccounts(store, store, &fakeQuota{exhausted: map[string]bool{}}, gateway, slog.New(slog.DiscardHandler))
+	alice := registerAccount(t, a, s, "alice", "correct horse")
+	login := func() {
+		t.Helper()
+		guest := createMember(t, s)
+		if _, _, err := a.Login(t.Context(), guest.Member, guest.MemberToken, clientIP, "alice", "correct horse"); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// 每一步之前都先登录一次，让这一步有会话可删。
+	for _, step := range []struct {
+		name string
+		op   func() error
+	}{
+		{"ChangePassword", func() error {
+			return a.ChangePassword(t.Context(), alice.Member.ID, alice.MemberToken, "correct horse", "correct horse")
+		}},
+		{"ResetPassword", func() error {
+			_, err := a.ResetPassword(t.Context(), "alice")
+			return err
+		}},
+	} {
+		login()
+		if err := step.op(); !errors.Is(err, realtime.ErrUnavailable) {
+			t.Errorf("广播失败时 %s = %v，期望 realtime.ErrUnavailable", step.name, err)
+		}
+	}
+}
+
+// 登录和退出撤销的是发起请求的这台设备的会话，页面随后会重新加载：广播失败时只记日志，照常成功。
+func TestLoginAndLogoutSucceedWhenRevocationFails(t *testing.T) {
+	store := porttest.New()
+	node := porttest.NewRedis().Node("a")
+	gateway := newGateway(store, node, noRevocation{node})
+	s := New(store, store, store, gateway)
+	var logs bytes.Buffer
+	a := NewAccounts(store, store, &fakeQuota{exhausted: map[string]bool{}}, gateway, slog.New(slog.NewJSONHandler(&logs, nil)))
+	alice := registerAccount(t, a, s, "alice", "correct horse")
+
+	guest := createMember(t, s)
+	session, _, err := a.Login(t.Context(), guest.Member, guest.MemberToken, clientIP, "alice", "correct horse")
+	if err != nil {
+		t.Fatalf("广播失败时 Login = %v，期望成功", err)
+	}
+	if _, ok := tokenOwner(t, store, guest.MemberToken); ok {
+		t.Error("游客原来的会话应已经删除")
+	}
+	if m, ok := tokenOwner(t, store, session.MemberToken); !ok || m.ID != alice.Member.ID {
+		t.Error("新会话应已经签发")
+	}
+	if err := a.Logout(t.Context(), session.MemberToken); err != nil {
+		t.Fatalf("广播失败时 Logout = %v，期望成功", err)
+	}
+	if _, ok := tokenOwner(t, store, session.MemberToken); ok {
+		t.Error("退出的会话应已经删除")
+	}
+	if n := bytes.Count(logs.Bytes(), []byte(`"level":"WARN"`)); n != 2 {
+		t.Errorf("警告日志 %d 条，期望 Login 和 Logout 各 1 条：\n%s", n, logs.String())
+	}
+}
+
 // 在登录校验密码之后、签发新会话之前修改账号的密码。
 type passwordChangedMembers struct {
 	*porttest.Store
 }
 
 func (s passwordChangedMembers) ReplaceSession(ctx context.Context, oldTokenHash []byte, memberID string, newTokenHash []byte, passwordHash string) error {
-	if err := s.SetPassword(ctx, memberID, "changed", nil); err != nil {
+	if _, err := s.SetPassword(ctx, memberID, "changed", nil); err != nil {
 		return err
 	}
 	return s.Store.ReplaceSession(ctx, oldTokenHash, memberID, newTokenHash, passwordHash)
 }
 
 func TestLoginRejectedWhenGuestSessionOrPasswordChangesMeanwhile(t *testing.T) {
-	a, s, store, _ := newAccounts(t)
+	a, s, store, _, _ := newAccounts(t)
 	registerAccount(t, a, s, "alice", "correct horse")
 	guest := createMember(t, s)
 	if err := store.DeleteSession(t.Context(), hashSecret(guest.MemberToken)); err != nil {
@@ -315,7 +480,8 @@ func TestLoginRejectedWhenGuestSessionOrPasswordChangesMeanwhile(t *testing.T) {
 		t.Errorf("游客的会话在校验之后失效，Login = %v，期望 ErrInvalidMemberToken", err)
 	}
 
-	changing := NewAccounts(passwordChangedMembers{store}, store, &fakeQuota{exhausted: map[string]bool{}})
+	node := porttest.NewRedis().Node("a")
+	changing := NewAccounts(passwordChangedMembers{store}, store, &fakeQuota{exhausted: map[string]bool{}}, newGateway(store, node, node), slog.New(slog.DiscardHandler))
 	guest = createMember(t, s)
 	if _, _, err := changing.Login(t.Context(), guest.Member, guest.MemberToken, clientIP, "alice", "correct horse"); !errors.Is(err, ErrInvalidCredentials) {
 		t.Errorf("密码在校验之后被修改，Login = %v，期望 ErrInvalidCredentials", err)
@@ -327,7 +493,7 @@ func TestLoginRejectedWhenGuestSessionOrPasswordChangesMeanwhile(t *testing.T) {
 
 // 名额占满、请求已经取消时，要做密码计算的请求以 context.Canceled 结束，不做的直接返回结果。
 func TestLoginDoesPasswordWorkForUnknownUsername(t *testing.T) {
-	a, s, _, _ := newAccounts(t)
+	a, s, _, _, _ := newAccounts(t)
 	registerAccount(t, a, s, "alice", "correct horse")
 	guest := createMember(t, s)
 	if _, err := dummyPasswordHash(); err != nil {

internal/app/identity/identity.go

@@ -84,7 +84,8 @@ func (s *Service) UpdateProfile(ctx context.Context, memberID string, p Profile)
 	return m, nil
 }
 
-func (s *Service) IssueTicket(ctx context.Context, memberID, roomID, password string) (string, error) {
+// memberToken 是发出请求的会话，票据属于这个会话：会话删除时票据随之失效。
+func (s *Service) IssueTicket(ctx context.Context, memberID, memberToken, roomID, password string) (string, error) {
 	r, err := s.rooms.FindRoom(ctx, roomID)
 	if err != nil {
 		return "", err
@@ -99,11 +100,15 @@ func (s *Service) IssueTicket(ctx context.Context, memberID, roomID, password st
 		}
 	}
 	ticket, hash := newSecret()
-	err = s.tickets.CreateTicket(ctx, port.NewTicket{TokenHash: hash, RoomID: r.Info.ID, MemberID: memberID, PasswordHash: r.PasswordHash})
+	err = s.tickets.CreateTicket(ctx, port.NewTicket{TokenHash: hash, RoomID: r.Info.ID, MemberID: memberID, PasswordHash: r.PasswordHash, SessionTokenHash: hashSecret(memberToken)})
 	if errors.Is(err, port.ErrPasswordChanged) {
 		// 刚才是按旧密码校验的；客户端重新输入密码后，按新密码校验。
 		return "", ErrWrongPassword
 	}
+	if errors.Is(err, port.ErrSessionNotFound) {
+		// 会话在鉴权之后被删除（例如另一台设备退出登录或修改了密码），此时成员 token 确实已经失效。
+		return "", ErrInvalidMemberToken
+	}
 	if err != nil {
 		return "", err
 	}
@@ -137,7 +142,16 @@ func (s *Service) EnterRoom(ctx context.Context, roomID, ticket string) (*realti
 	if holder.Bot {
 		return nil, ErrBotSession
 	}
-	return s.gateway.Join(logctx.WithMember(logctx.WithRoom(ctx, r.Info.ID), holder.Member.ID), r.Info.ID, holder.Member)
+	session, err := s.gateway.Join(logctx.WithMember(logctx.WithRoom(ctx, r.Info.ID), holder.Member.ID), r.Info.ID, holder.Member, holder.SessionTokenHash)
+	if err != nil {
+		return nil, err
+	}
+	// 作废时先删除票据、再通知断开：会话加入网关之前发出的通知它收不到，所以加入之后再确认一次票据。
+	if _, err := s.VerifyTicket(ctx, ticket); err != nil {
+		s.gateway.Leave(session)
+		return nil, err
+	}
+	return session, nil
 }
 
 func NewAPIKey() (string, port.NewAPIKey) {

internal/app/identity/identity_test.go

@@ -9,6 +9,7 @@ import (
 	"log/slog"
 	"regexp"
 	"testing"
+	"testing/synctest"
 
 	"github.com/dihemoe/alisten/internal/app/port"
 	"github.com/dihemoe/alisten/internal/app/port/porttest"
@@ -152,14 +153,88 @@ func TestIssueTicketFailsWhenPasswordChangesMeanwhile(t *testing.T) {
 	changed := passwordChangedStore{Store: store, newHash: hashPassword(t, "新密码")}
 	node := porttest.NewRedis().Node("a")
 	s := New(store, store, changed, newGateway(store, node, node))
-	alice := createMember(t, s).Member
+	alice := createMember(t, s)
 	id := createRoom(t, store, "旧密码")
 
-	if _, err := s.IssueTicket(t.Context(), alice.ID, id, "旧密码"); !errors.Is(err, ErrWrongPassword) {
+	if _, err := s.IssueTicket(t.Context(), alice.Member.ID, alice.MemberToken, id, "旧密码"); !errors.Is(err, ErrWrongPassword) {
 		t.Errorf("密码在校验之后被修改，错误 = %v，期望 ErrWrongPassword", err)
 	}
 }
 
+func TestTicketBelongsToIssuingSession(t *testing.T) {
+	s, store := newService(t)
+	alice := createMember(t, s)
+	id := createRoom(t, store, "")
+	ticket, err := s.IssueTicket(t.Context(), alice.Member.ID, alice.MemberToken, id, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	holder, err := s.VerifyTicket(t.Context(), ticket)
+	if err != nil || holder.Member != alice.Member || !bytes.Equal(holder.SessionTokenHash, hashSecret(alice.MemberToken)) {
+		t.Errorf("票据的持有者 = %#v, %v，期望属于 alice 签发它的会话", holder, err)
+	}
+}
+
+func TestIssueTicketRejectsDeletedSession(t *testing.T) {
+	s, store := newService(t)
+	alice := createMember(t, s)
+	id := createRoom(t, store, "")
+	if err := store.DeleteSession(t.Context(), hashSecret(alice.MemberToken)); err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := s.IssueTicket(t.Context(), alice.Member.ID, alice.MemberToken, id, ""); !errors.Is(err, ErrInvalidMemberToken) {
+		t.Errorf("会话在签发之前被删除，错误 = %v，期望 ErrInvalidMemberToken", err)
+	}
+}
+
+// 第一次查询照常，之后的查询都认为票据已经作废。
+type revokedAfterFirstLookup struct {
+	port.Tickets
+	lookups int
+}
+
+func (r *revokedAfterFirstLookup) TicketHolder(ctx context.Context, tokenHash []byte) (port.TicketHolder, error) {
+	r.lookups++
+	if r.lookups > 1 {
+		return port.TicketHolder{}, port.ErrTicketNotFound
+	}
+	return r.Tickets.TicketHolder(ctx, tokenHash)
+}
+
+func TestEnterRoomLeavesWhenTicketIsRevokedWhileJoining(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		store := porttest.New()
+		node := porttest.NewRedis().Node("a")
+		host, gateway := realtime.New(store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler))
+		ctx, stop := context.WithCancel(t.Context())
+		wait, err := realtime.Start(ctx, host, gateway)
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(func() {
+			_ = host.Shutdown(context.Background())
+			_ = gateway.Shutdown(context.Background())
+			stop()
+			wait()
+		})
+		s := New(store, store, &revokedAfterFirstLookup{Tickets: store}, gateway)
+		alice := createMember(t, s)
+		id := createRoom(t, store, "")
+		ticket, err := s.IssueTicket(t.Context(), alice.Member.ID, alice.MemberToken, id, "")
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if _, err := s.EnterRoom(t.Context(), id, ticket); !errors.Is(err, ErrInvalidTicket) {
+			t.Errorf("加入期间票据被作废，错误 = %v，期望 ErrInvalidTicket", err)
+		}
+		if n := gateway.Sessions(); n != 0 {
+			t.Errorf("网关上残留 %d 个会话，期望 0", n)
+		}
+	})
+}
+
 type loggedSubscriptions struct {
 	port.Bus
 	logger *slog.Logger
@@ -177,12 +252,13 @@ func TestEnterRoomJoinsTheStoredRoom(t *testing.T) {
 	node := porttest.NewRedis().Node("a")
 	var logs bytes.Buffer
 	s := New(store, store, store, newGateway(store, node, loggedSubscriptions{node, slog.New(logctx.NewHandler(slog.NewJSONHandler(&logs, nil)))}))
-	alice := createMember(t, s).Member
+	session := createMember(t, s)
+	alice := session.Member
 	id, err := store.CreateRoom(t.Context(), port.NewRoom{Kind: room.Persistent, Name: "大厅", Slug: "hall"})
 	if err != nil {
 		t.Fatal(err)
 	}
-	ticket, err := s.IssueTicket(t.Context(), alice.ID, id, "")
+	ticket, err := s.IssueTicket(t.Context(), alice.ID, session.MemberToken, id, "")
 	if err != nil {
 		t.Fatal(err)
 	}

internal/app/lobby/lobby_test.go

@@ -47,7 +47,7 @@ func TestOnlineCount(t *testing.T) {
 	// 同一个成员的两个会话算两个在线。
 	alice := member.Member{ID: "00000000-0000-7000-8000-00000000000a", Name: "alice"}
 	for range 2 {
-		session, err := gateway.Join(t.Context(), busy, alice)
+		session, err := gateway.Join(t.Context(), busy, alice, nil)
 		if err != nil {
 			t.Fatal(err)
 		}

internal/app/port/bus.go

@@ -92,19 +92,25 @@ type (
 		Outcome PrepareOutcome
 	}
 
+	// 这些会话已经删除，用它们签发的票据建立的会话和事件流要以 4401 断开。
+	SessionsRevokedDelivery struct {
+		SessionTokenHashes [][]byte
+	}
+
 	// 订阅连接曾经断开，断开期间发给本实例的消息都已丢失。
 	SubscriptionLost struct{}
 )
 
-func (RoomDelivery) delivery()     {}
-func (InputDelivery) delivery()    {}
-func (MemberDelivery) delivery()   {}
-func (PreparedDelivery) delivery() {}
-func (SubscriptionLost) delivery() {}
+func (RoomDelivery) delivery()            {}
+func (InputDelivery) delivery()           {}
+func (MemberDelivery) delivery()          {}
+func (PreparedDelivery) delivery()        {}
+func (SessionsRevokedDelivery) delivery() {}
+func (SubscriptionLost) delivery()        {}
 
 // 实现绑定在本实例上：订阅的是本实例的频道，发出的消息带有本实例的 ID。
 type Bus interface {
-	// 订阅本实例的频道、成员资料的频道和准备结果的频道，返回时订阅已经生效。
+	// 订阅本实例的频道、成员的频道和准备结果的频道，返回时订阅已经生效。
 	SubscribeInstance(ctx context.Context) error
 	// 返回时订阅已经生效：此后发布到这个房间频道的消息都会收到。
 	SubscribeRoom(ctx context.Context, roomID string) error
@@ -114,6 +120,8 @@ type Bus interface {
 	SendInput(ctx context.Context, instanceID string, in RoomInput) (bool, error)
 	PublishMember(ctx context.Context, m member.Member) error
 	// 各实例都会收到，包括发布的实例。
+	PublishSessionsRevoked(ctx context.Context, sessionTokenHashes [][]byte) error
+	// 各实例都会收到，包括发布的实例。
 	PublishPrepared(ctx context.Context, o PrepareOutcome) error
 	// 按收到的顺序返回下一条消息，直到 ctx 结束；ctx 结束之后 Bus 不能再使用。只能由一个 goroutine 调用。
 	Receive(ctx context.Context) (Delivery, error)

internal/app/port/member.go

@@ -30,6 +30,7 @@ type AccountProfile struct {
 }
 
 // 成员 token 只以 SHA-256 哈希的形式出现在端口中。一个成员 token 是成员的一个会话，同一个成员可以有多个会话。
+// 删除会话时，签发给这个会话的票据一并删除。
 type Members interface {
 	// 创建游客和它的第一个会话。
 	CreateMember(ctx context.Context, tokenHash []byte, name string) (member.Member, error)
@@ -46,6 +47,6 @@ type Members interface {
 	ReplaceSession(ctx context.Context, oldTokenHash []byte, memberID string, newTokenHash []byte, passwordHash string) error
 	DeleteSession(ctx context.Context, tokenHash []byte) error
 	// 在同一个事务中修改已注册成员的密码哈希，并删除该成员除 keepTokenHash 以外的会话，keepTokenHash 为 nil 时全部删除。
-	// 成员是游客时返回 ErrMemberNotFound。
-	SetPassword(ctx context.Context, memberID, passwordHash string, keepTokenHash []byte) error
+	// 返回被删除的会话的 token 哈希。成员是游客时返回 ErrMemberNotFound。
+	SetPassword(ctx context.Context, memberID, passwordHash string, keepTokenHash []byte) ([][]byte, error)
 }

internal/app/port/porttest/redis.go

@@ -282,6 +282,11 @@ func (n *Node) PublishMember(_ context.Context, m member.Member) error {
 	return err
 }
 
+func (n *Node) PublishSessionsRevoked(_ context.Context, sessionTokenHashes [][]byte) error {
+	_, err := n.publish("members", port.SessionsRevokedDelivery{SessionTokenHashes: sessionTokenHashes})
+	return err
+}
+
 func (n *Node) PublishPrepared(_ context.Context, o port.PrepareOutcome) error {
 	_, err := n.publish("tracks.prepared", port.PreparedDelivery{Outcome: o})
 	return err

internal/app/port/porttest/store.go

@@ -45,6 +45,7 @@ type storedMember struct {
 type storedTicket struct {
 	roomID    string
 	memberID  string
+	session   string
 	createdAt time.Time
 }
 
@@ -203,7 +204,7 @@ func (s *Store) ReplaceSession(_ context.Context, oldTokenHash []byte, memberID
 	if _, ok := s.sessions[string(newTokenHash)]; ok {
 		return errors.New("token 哈希重复")
 	}
-	delete(s.sessions, string(oldTokenHash))
+	s.deleteSession(string(oldTokenHash))
 	s.sessions[string(newTokenHash)] = memberID
 	return nil
 }
@@ -211,27 +212,39 @@ func (s *Store) ReplaceSession(_ context.Context, oldTokenHash []byte, memberID
 func (s *Store) DeleteSession(_ context.Context, tokenHash []byte) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	delete(s.sessions, string(tokenHash))
+	s.deleteSession(string(tokenHash))
 	return nil
 }
 
-func (s *Store) SetPassword(_ context.Context, memberID, passwordHash string, keepTokenHash []byte) error {
+func (s *Store) SetPassword(_ context.Context, memberID, passwordHash string, keepTokenHash []byte) ([][]byte, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	m := s.member(memberID)
 	if m == nil || m.username == "" {
-		return port.ErrMemberNotFound
+		return nil, port.ErrMemberNotFound
 	}
 	m.passwordHash = passwordHash
-	s.deleteSessions(memberID, keepTokenHash)
-	return nil
+	return s.deleteSessions(memberID, keepTokenHash), nil
 }
 
 // 调用方持有 s.mu。
-func (s *Store) deleteSessions(memberID string, keepTokenHash []byte) {
-	maps.DeleteFunc(s.sessions, func(hash, id string) bool {
-		return id == memberID && (keepTokenHash == nil || hash != string(keepTokenHash))
-	})
+func (s *Store) deleteSessions(memberID string, keepTokenHash []byte) [][]byte {
+	var deleted [][]byte
+	for hash, id := range s.sessions {
+		if id == memberID && (keepTokenHash == nil || hash != string(keepTokenHash)) {
+			deleted = append(deleted, []byte(hash))
+		}
+	}
+	for _, hash := range deleted {
+		s.deleteSession(string(hash))
+	}
+	return deleted
+}
+
+// 与 postgres 的外键一致：会话的票据随会话一起删除。调用方持有 s.mu。
+func (s *Store) deleteSession(hash string) {
+	delete(s.sessions, hash)
+	maps.DeleteFunc(s.tickets, func(_ string, t storedTicket) bool { return t.session == hash })
 }
 
 func (s *Store) CreateTicket(_ context.Context, t port.NewTicket) error {
@@ -247,10 +260,13 @@ func (s *Store) CreateTicket(_ context.Context, t port.NewTicket) error {
 	if !slices.ContainsFunc(s.members, func(m storedMember) bool { return m.member.ID == t.MemberID }) {
 		return errors.New("成员不存在")
 	}
+	if _, ok := s.sessions[string(t.SessionTokenHash)]; t.SessionTokenHash != nil && !ok {
+		return port.ErrSessionNotFound
+	}
 	if _, ok := s.tickets[string(t.TokenHash)]; ok {
 		return errors.New("票据哈希重复")
 	}
-	s.tickets[string(t.TokenHash)] = storedTicket{roomID: r.Info.ID, memberID: t.MemberID, createdAt: time.Now()}
+	s.tickets[string(t.TokenHash)] = storedTicket{roomID: r.Info.ID, memberID: t.MemberID, session: string(t.SessionTokenHash), createdAt: time.Now()}
 	return nil
 }
 
@@ -263,7 +279,11 @@ func (s *Store) TicketHolder(_ context.Context, tokenHash []byte) (port.TicketHo
 	}
 	for _, m := range s.members {
 		if m.member.ID == t.memberID {
-			return port.TicketHolder{RoomID: t.roomID, Member: m.member, Bot: s.bots[t.roomID] == t.memberID}, nil
+			holder := port.TicketHolder{RoomID: t.roomID, Member: m.member, Bot: s.bots[t.roomID] == t.memberID}
+			if t.session != "" {
+				holder.SessionTokenHash = []byte(t.session)
+			}
+			return holder, nil
 		}
 	}
 	return port.TicketHolder{}, port.ErrTicketNotFound

internal/app/port/ticket.go

@@ -18,6 +18,8 @@ type NewTicket struct {
 	MemberID  string
 	// 签发前校验密码时依据的密码哈希，空字符串表示房间没有密码。
 	PasswordHash string
+	// 发出请求的会话的 token 哈希：会话删除时票据随之失效。
+	SessionTokenHash []byte
 }
 
 type TicketHolder struct {
@@ -26,11 +28,13 @@ type TicketHolder struct {
 	Member member.Member
 	// 票据是常驻房间的 API Key，持有者是这个房间的 bot 成员。
 	Bot bool
+	// 票据所属的会话的 token 哈希，API Key 没有会话，为 nil。
+	SessionTokenHash []byte
 }
 
 type Tickets interface {
 	// 房间的密码哈希已经不是 t.PasswordHash 时返回 ErrPasswordChanged：密码在校验之后被修改过，
-	// 设置或修改密码时的作废已经执行，此时签发的票据不会再被作废。
+	// 设置或修改密码时的作废已经执行，此时签发的票据不会再被作废。t.SessionTokenHash 的会话已经不存在时返回 ErrSessionNotFound。
 	CreateTicket(ctx context.Context, t NewTicket) error
 	TicketHolder(ctx context.Context, tokenHash []byte) (TicketHolder, error)
 }

internal/app/realtime/cluster_test.go

@@ -398,7 +398,7 @@ func TestGatewayRejoinsAfterItsAliveKeyExpired(t *testing.T) {
 		if _, err := a.gateway.Do(t.Context(), id, alice, room.Sync{}); err != nil {
 			t.Fatal(err)
 		}
-		s, err := g.Join(t.Context(), id, bob)
+		s, err := g.Join(t.Context(), id, bob, nil)
 		if err != nil {
 			t.Fatal(err)
 		}

internal/app/realtime/gateway.go

@@ -105,12 +105,12 @@ func New(snapshots port.RoomSnapshots, directory port.RoomDirectory, music *cata
 }
 
 // 会话结束时必须调用 Leave。
-func (g *Gateway) Join(ctx context.Context, roomID string, m member.Member) (*Session, error) {
+func (g *Gateway) Join(ctx context.Context, roomID string, m member.Member, memberSession []byte) (*Session, error) {
 	link, err := g.hold(ctx, roomID)
 	if err != nil {
 		return nil, err
 	}
-	s := newSession(room.SessionID(g.nextID("s")), roomID, m)
+	s := newSession(room.SessionID(g.nextID("s")), roomID, m, memberSession)
 	s.link = link
 	if !g.register(s) {
 		g.unhold(roomID, link, false)
@@ -208,6 +208,19 @@ func (g *Gateway) RevokeAPIKey(ctx context.Context, roomID string) error {
 	return g.announce(ctx, roomID, port.APIKeyReset{})
 }
 
+// RevokeSessions 在成员会话删除之后通知各实例，以 CloseRevoked 关闭用这些会话的票据建立的会话和事件流。
+func (g *Gateway) RevokeSessions(ctx context.Context, memberSessions [][]byte) error {
+	if len(memberSessions) == 0 {
+		return nil
+	}
+	ctx, cancel := context.WithTimeout(ctx, clusterTimeout)
+	defer cancel()
+	if err := g.bus.PublishSessionsRevoked(ctx, memberSessions); err != nil {
+		return unavailable(err)
+	}
+	return nil
+}
+
 func (g *Gateway) announce(ctx context.Context, roomID string, m port.RoomMessage) error {
 	ctx, cancel := context.WithTimeout(ctx, clusterTimeout)
 	defer cancel()
@@ -470,6 +483,34 @@ func (g *Gateway) memberChanged(m member.Member) {
 	}()
 }
 
+// 在接收消息的 goroutine 上调用，不能阻塞。
+func (g *Gateway) sessionsRevoked(memberSessions [][]byte) {
+	revoked := map[string]bool{}
+	for _, h := range memberSessions {
+		revoked[string(h)] = true
+	}
+	g.mu.Lock()
+	var closing []*sendQueue
+	for _, set := range g.sessions {
+		for s := range set {
+			if revoked[s.memberSession] {
+				closing = append(closing, s.sendQueue)
+			}
+		}
+	}
+	for _, set := range g.observers {
+		for _, o := range set {
+			if revoked[o.memberSession] {
+				closing = append(closing, o.sendQueue)
+			}
+		}
+	}
+	g.mu.Unlock()
+	for _, q := range closing {
+		q.close(CloseRevoked)
+	}
+}
+
 // owner 靠 Join、Leave 和 MemberUpdated 知道谁在线，这几条消息丢失后没有别的办法补回：离开的会话会一直留在在线列表中，
 // 临时房间也因此无法关闭；仍然在线的会话则不在列表中。所以送不到 owner 时要一直重试，直到送达。
 // next 在每次尝试前调用，返回这一次要发的输入；返回 false 表示这条消息已经不必再发。
@@ -628,6 +669,8 @@ func (g *Gateway) receive(ctx context.Context, lost chan<- struct{}) {
 			g.host.deliver(d.Context, d.Input)
 		case port.MemberDelivery:
 			g.memberChanged(d.Member)
+		case port.SessionsRevokedDelivery:
+			g.sessionsRevoked(d.SessionTokenHashes)
 		case port.PreparedDelivery:
 			g.host.preparer.Finished(d.Outcome)
 		case port.SubscriptionLost:

internal/app/realtime/gateway_test.go

@@ -57,7 +57,7 @@ func startLoggingInstance(t *testing.T, redis *porttest.Redis, id string, snapsh
 
 func (in *instance) join(t *testing.T, roomID string, m member.Member) *Session {
 	t.Helper()
-	s, err := in.gateway.Join(t.Context(), roomID, m)
+	s, err := in.gateway.Join(t.Context(), roomID, m, nil)
 	if err != nil {
 		t.Fatalf("%s 加入房间：%v", m.Name, err)
 	}
@@ -366,7 +366,7 @@ func TestShutdownClosesSessions(t *testing.T) {
 		store := porttest.New()
 		f := newGatewayFixture(t, store, store)
 		id := createRoom(t, store, room.Persistent)
-		a, err := f.gateway.Join(t.Context(), id, alice)
+		a, err := f.gateway.Join(t.Context(), id, alice, nil)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -384,7 +384,7 @@ func TestShutdownClosesSessions(t *testing.T) {
 		if err := <-done; err != nil {
 			t.Fatal(err)
 		}
-		if _, err := f.gateway.Join(t.Context(), id, bob); !errors.Is(err, ErrUnavailable) {
+		if _, err := f.gateway.Join(t.Context(), id, bob, nil); !errors.Is(err, ErrUnavailable) {
 			t.Errorf("Shutdown 之后加入的错误 = %v，期望 ErrUnavailable", err)
 		}
 	})
@@ -463,7 +463,7 @@ func TestUnsubscribeAfterLastUse(t *testing.T) {
 		sessionOnly := createRoom(t, store, room.Persistent)
 		withHTTP := createRoom(t, store, room.Persistent)
 
-		s, err := f.gateway.Join(t.Context(), sessionOnly, alice)
+		s, err := f.gateway.Join(t.Context(), sessionOnly, alice, nil)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -481,7 +481,7 @@ func TestUnsubscribeAfterLastUse(t *testing.T) {
 			t.Fatal(err)
 		}
 		time.Sleep(30 * time.Second)
-		s, err = f.gateway.Join(t.Context(), withHTTP, alice)
+		s, err = f.gateway.Join(t.Context(), withHTTP, alice, nil)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -620,7 +620,7 @@ func TestConfirmedSubscriptionIsNotRepeated(t *testing.T) {
 		store := porttest.New()
 		g, node, bus := startCountingGateway(t, store)
 		id := createRoom(t, store, room.Persistent)
-		s, err := g.Join(t.Context(), id, alice)
+		s, err := g.Join(t.Context(), id, alice, nil)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -657,7 +657,7 @@ func TestConfirmationAcrossDisconnectIsNotReused(t *testing.T) {
 		store := porttest.New()
 		g, node, bus := startCountingGateway(t, store)
 		id := createRoom(t, store, room.Persistent)
-		s, err := g.Join(t.Context(), id, alice)
+		s, err := g.Join(t.Context(), id, alice, nil)
 		if err != nil {
 			t.Fatal(err)
 		}

internal/app/realtime/observer.go

@@ -17,26 +17,28 @@ var ErrTooManyObservers = errors.New("这个房间的事件流超过 5 条，请
 // Observer 是房间的事件流：先收到房间的快照，之后是房间的事件；房间开始新的一次运行时事件重新编号，再收到一份新的快照。
 type Observer struct {
 	*sendQueue
-	room       string
-	link       *roomLink
-	member     member.Member
-	bot        bool
-	credential string
+	room   string
+	link   *roomLink
+	member member.Member
+	// 凭证签发给的成员会话的 token 哈希，API Key 没有会话，为空。
+	memberSession string
+	bot           bool
+	credential    string
 
 	// 由 sendQueue 的 mu 保护：最近一次快照的任期，和收到过的 owner 变更通知中最大的任期。
 	term, latest uint32
 }
 
 // Observe 订阅房间的事件但不加入房间：事件流不出现在在线列表中，不计入投票切歌所需的人数，也不让 AutoDJ 认为房间里有人。
-// m 是凭证的持有者，bot 表示凭证是常驻房间的 API Key，credential 是凭证的哈希。名额用完时，同一凭证打开的最早的事件流以
-// CloseReplaced 关闭，名额让给这一条；没有同一凭证的事件流时返回 ErrTooManyObservers。
+// m 是凭证的持有者，memberSession 是凭证签发给的成员会话，bot 表示凭证是常驻房间的 API Key，credential 是凭证的哈希。
+// 名额用完时，同一凭证打开的最早的事件流以 CloseReplaced 关闭，名额让给这一条；没有同一凭证的事件流时返回 ErrTooManyObservers。
 // 返回时快照已经放进队列。结束时必须调用一次 Unobserve。
-func (g *Gateway) Observe(ctx context.Context, roomID string, m member.Member, bot bool, credential string) (*Observer, error) {
+func (g *Gateway) Observe(ctx context.Context, roomID string, m member.Member, memberSession []byte, bot bool, credential string) (*Observer, error) {
 	link, err := g.hold(ctx, roomID)
 	if err != nil {
 		return nil, err
 	}
-	o := &Observer{sendQueue: newSendQueue(), room: roomID, link: link, member: m, bot: bot, credential: credential}
+	o := &Observer{sendQueue: newSendQueue(), room: roomID, link: link, member: m, memberSession: string(memberSession), bot: bot, credential: credential}
 	if err := g.watch(o); err != nil {
 		g.unhold(roomID, link, false)
 		return nil, err

internal/app/realtime/observer_test.go

@@ -25,7 +25,7 @@ var (
 // 只能在 synctest 的 bubble 中调用。
 func (in *instance) observe(t *testing.T, roomID string, m member.Member, bot bool, credential string) *Observer {
 	t.Helper()
-	o, err := in.gateway.Observe(t.Context(), roomID, m, bot, credential)
+	o, err := in.gateway.Observe(t.Context(), roomID, m, nil, bot, credential)
 	if err != nil {
 		t.Fatalf("%s 打开事件流：%v", m.Name, err)
 	}
@@ -126,6 +126,59 @@ func TestAPIKeyResetClosesOnlyBotObservers(t *testing.T) {
 	})
 }
 
+func TestRevokedSessionsCloseOnlyTheirConnections(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		store := porttest.New()
+		f := newClusterFixture(t, store, store)
+		lounge := createRoom(t, store, room.Persistent)
+		join := func(in *instance, roomID string, m member.Member, memberSession string) *Session {
+			t.Helper()
+			s, err := in.gateway.Join(t.Context(), roomID, m, []byte(memberSession))
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Cleanup(func() { in.gateway.Leave(s) })
+			return s
+		}
+		observe := func(in *instance, m member.Member, memberSession []byte, bot bool, credential string) *Observer {
+			t.Helper()
+			o, err := in.gateway.Observe(t.Context(), f.room, m, memberSession, bot, credential)
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Cleanup(func() { in.gateway.Unobserve(o) })
+			return o
+		}
+		revokedOnA, revokedOnB, kept := join(f.a, f.room, alice, "alice-1"), join(f.b, lounge, alice, "alice-1"), join(f.b, f.room, alice, "alice-2")
+		other := join(f.b, f.room, bob, "bob-1")
+		revokedStream, keptStream := observe(f.b, alice, []byte("alice-1"), false, "t1"), observe(f.b, alice, []byte("alice-2"), false, "t2")
+		bot := observe(f.b, robot, nil, true, "key")
+		for _, q := range []interface {
+			Next(context.Context) (Message, CloseReason, error)
+		}{revokedOnA, revokedOnB, kept, other, revokedStream, keptStream, bot} {
+			received(t, q)
+		}
+
+		if err := f.a.gateway.RevokeSessions(t.Context(), [][]byte{[]byte("alice-1"), []byte("carol-1")}); err != nil {
+			t.Fatal(err)
+		}
+		revoked := fmt.Sprintf("closed %d", CloseRevoked)
+		checkReceived(t, revokedOnA, revoked)
+		checkReceived(t, revokedOnB, revoked)
+		checkObserved(t, revokedStream, revoked)
+		for _, s := range []*Session{kept, other} {
+			if got := received(t, s); slices.Contains(got, revoked) {
+				t.Errorf("%s 的会话也被关闭了：%v", s.Member().Name, got)
+			}
+		}
+		for _, o := range []*Observer{keptStream, bot} {
+			if got := received(t, o); slices.Contains(got, revoked) {
+				t.Errorf("%s 的事件流也被关闭了：%v", o.member.Name, got)
+			}
+		}
+	})
+}
+
 func TestObserverGetsNewSnapshotWhenRunChanges(t *testing.T) {
 	synctest.Test(t, func(t *testing.T) {
 		store := &failingStore{Store: porttest.New()}
@@ -203,7 +256,7 @@ func TestObserverSkipsSnapshotOfEarlierRun(t *testing.T) {
 		opened := make(chan error, 1)
 		go func() {
 			var err error
-			o, err = b.gateway.Observe(context.Background(), id, robot, true, "key")
+			o, err = b.gateway.Observe(context.Background(), id, robot, nil, true, "key")
 			opened <- err
 		}()
 		synctest.Wait()
@@ -225,14 +278,14 @@ func TestObserverLimitPerRoom(t *testing.T) {
 		store := porttest.New()
 		f := newGatewayFixture(t, store, store)
 		full, other := createRoom(t, store, room.Persistent), createRoom(t, store, room.Persistent)
-		first, err := f.gateway.Observe(t.Context(), full, robot, true, "key")
+		first, err := f.gateway.Observe(t.Context(), full, robot, nil, true, "key")
 		if err != nil {
 			t.Fatal(err)
 		}
 		for range observerLimit - 1 {
 			f.observe(t, full, robot, true, "key")
 		}
-		if _, err := f.gateway.Observe(t.Context(), full, alice, false, "alice"); !errors.Is(err, ErrTooManyObservers) {
+		if _, err := f.gateway.Observe(t.Context(), full, alice, nil, false, "alice"); !errors.Is(err, ErrTooManyObservers) {
 			t.Errorf("第 %d 个事件流的错误 = %v，期望 ErrTooManyObservers", observerLimit+1, err)
 		}
 		f.observe(t, other, alice, false, "alice")
@@ -258,7 +311,7 @@ func TestFullRoomReplacesOldestObserverOfSameCredential(t *testing.T) {
 			checkObserved(t, o, "snapshot 1/0 []")
 		}
 
-		if _, err := f.gateway.Observe(t.Context(), id, bob, false, "bob 的另一张票据"); !errors.Is(err, ErrTooManyObservers) {
+		if _, err := f.gateway.Observe(t.Context(), id, bob, nil, false, "bob 的另一张票据"); !errors.Is(err, ErrTooManyObservers) {
 			t.Errorf("名额用完、没有同一凭证的事件流时的错误 = %v，期望 ErrTooManyObservers", err)
 		}
 		for _, o := range open {
@@ -273,7 +326,7 @@ func TestFailedObserveFreesItsSlot(t *testing.T) {
 		f := newGatewayFixture(t, store, store)
 		missing := "01a15778-1911-7a28-8b2e-e08c2a25f388"
 		for i := range observerLimit + 1 {
-			if _, err := f.gateway.Observe(t.Context(), missing, robot, true, fmt.Sprint(i)); !errors.Is(err, port.ErrRoomNotFound) {
+			if _, err := f.gateway.Observe(t.Context(), missing, robot, nil, true, fmt.Sprint(i)); !errors.Is(err, port.ErrRoomNotFound) {
 				t.Fatalf("第 %d 次打开不存在的房间的事件流 = %v，期望 ErrRoomNotFound", i+1, err)
 			}
 		}

internal/app/realtime/session.go

@@ -83,6 +83,8 @@ type Session struct {
 	id   room.SessionID
 	room string
 	link *roomLink
+	// 连接所用的票据签发给的成员会话的 token 哈希。
+	memberSession string
 	// 在发出 Leave 之前设置。重新加入的 Join 可能排在 Leave 之后，网关据此补发 Leave。
 	left atomic.Bool
 
@@ -91,8 +93,8 @@ type Session struct {
 	outstanding int
 }
 
-func newSession(id room.SessionID, roomID string, m member.Member) *Session {
-	return &Session{sendQueue: newSendQueue(), id: id, room: roomID, member: m}
+func newSession(id room.SessionID, roomID string, m member.Member, memberSession []byte) *Session {
+	return &Session{sendQueue: newSendQueue(), id: id, room: roomID, member: m, memberSession: string(memberSession)}
 }
 
 func (s *Session) ended() bool {

internal/bootstrap/public.go

@@ -78,7 +78,7 @@ func newPublic(cfg serveConfig, pool *pgxpool.Pool, rdb *goredis.Client, instanc
 	rooms := lobby.New(store, leases)
 	services := api.Services{
 		Identity: identity.New(store, store, store, gateway),
-		Accounts: identity.NewAccounts(store, store, redis.NewAccountQuota(rdb)),
+		Accounts: identity.NewAccounts(store, store, redis.NewAccountQuota(rdb), gateway, base.With("component", "accounts")),
 		Lobby:    rooms,
 		Admin:    admin.New(cfg.adminToken, rooms, store, store, store, src.neteaseLogin, src.qqLogin, src.bilibiliLogin, store, gateway),
 		Gateway:  gateway,

internal/bootstrap/testdata/http/events.txtar

@@ -107,7 +107,7 @@ Location: /api/admin/rooms/$hall/api-key
 {"key": "$resetKey"}
 -- event bot --
 event: close
-data: {"code": 4401, "message": "凭证已失效：房间设置或修改了密码，或者 API Key 已重置"}
+data: {"code": 4401, "message": "凭证已失效：房间设置或修改了密码，API Key 已重置，或者账号在这台设备上退出了登录、密码被修改或重置"}
 
 -- request --
 POST /api/room/messages

internal/bootstrap/testdata/ws/account-sessions.txtar

@@ -0,0 +1,142 @@
+票据属于签发它的会话：账号在两台设备上各自进入同一个房间，一台设备修改密码之后，另一台设备的 WebSocket 以 4401 断开，
+用它的票据打开的事件流收到 4401 的 close；这台设备的旧票据再也不能用来进房或调用房间接口。发出请求的设备不受影响。
+
+-- request --
+POST /api/members
+
+{}
+-- response --
+201
+{"memberToken": "$alice", "member": {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}}
+
+-- request --
+POST /api/account
+Authorization: Bearer $alice
+
+{"username": "alice", "password": "correct horse"}
+-- response --
+201
+{"account": {"username": "alice"}}
+
+-- request --
+POST /api/members
+
+{}
+-- response --
+201
+{"memberToken": "$guest", "member": {"id": "$guestId", "name": "$guestName", "avatarHash": ""}}
+
+-- request --
+POST /api/sessions
+Authorization: Bearer $guest
+
+{"username": "alice", "password": "correct horse"}
+-- response --
+201
+{"memberToken": "$alice2", "member": {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}, "account": {"username": "alice"}}
+
+-- request --
+POST /api/rooms
+
+{"name": "临时", "description": ""}
+-- response --
+201
+Location: /api/rooms/$room
+{"id": "$room"}
+
+-- request --
+POST /api/rooms/$room/tickets
+Authorization: Bearer $alice
+
+{}
+-- response --
+201
+{"ticket": "$ticketA"}
+
+-- request --
+POST /api/rooms/$room/tickets
+Authorization: Bearer $alice2
+
+{}
+-- response --
+201
+{"ticket": "$ticketB"}
+
+-- connect a --
+-- send a --
+{"kind": "request", "type": "hello", "id": "h1", "data": {"roomId": "$room", "ticket": "$ticketA", "protocolVersion": "$protocolVersion"}}
+-- receive a --
+{"kind": "response", "type": "hello", "id": "h1", "data": {
+  "member": {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""},
+  "serverTimeMs": 1,
+  "room": {"term": 1, "seq": 1, "room": {"id": "$room", "name": "临时", "description": "", "kind": "ephemeral"},
+    "playMode": "sequential", "nowPlaying": null, "pending": null, "items": [],
+    "members": {"sessions": [{"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}], "onlineCount": 1}}}}
+
+-- connect b --
+-- send b --
+{"kind": "request", "type": "hello", "id": "h1", "data": {"roomId": "$room", "ticket": "$ticketB", "protocolVersion": "$protocolVersion"}}
+-- receive b --
+{"kind": "response", "type": "hello", "id": "h1", "data": {
+  "member": {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""},
+  "serverTimeMs": 1,
+  "room": {"term": 1, "seq": 2, "room": {"id": "$room", "name": "临时", "description": "", "kind": "ephemeral"},
+    "playMode": "sequential", "nowPlaying": null, "pending": null, "items": [],
+    "members": {"sessions": [{"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}, {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}], "onlineCount": 2}}}}
+-- receive a --
+{"kind": "event", "type": "members.updated", "term": 1, "seq": 2, "data": {
+  "sessions": [{"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}, {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}], "onlineCount": 2}}
+
+-- stream b --
+Authorization: Bearer $ticketB
+-- event b --
+event: snapshot
+data: {"room": {"term": 1, "seq": 2, "room": {"id": "$room", "name": "临时", "description": "", "kind": "ephemeral"},
+  "playMode": "sequential", "nowPlaying": null, "pending": null, "items": [],
+  "members": {"sessions": [{"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}, {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}], "onlineCount": 2}}}
+
+-- request --
+PUT /api/account/password
+Authorization: Bearer $alice
+
+{"currentPassword": "correct horse", "newPassword": "battery staple"}
+-- response --
+204
+-- closed b --
+4401
+-- event b --
+event: close
+data: {"code": 4401, "message": "凭证已失效：房间设置或修改了密码，API Key 已重置，或者账号在这台设备上退出了登录、密码被修改或重置"}
+-- receive a --
+{"kind": "event", "type": "members.updated", "term": 1, "seq": 3, "data": {
+  "sessions": [{"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}], "onlineCount": 1}}
+
+-- send a --
+{"kind": "request", "type": "ping", "id": "p1", "data": {"clientTimeMs": 1234}}
+-- receive a --
+{"kind": "response", "type": "ping", "id": "p1", "data": {"clientTimeMs": 1234, "serverTimeMs": 1}}
+
+-- connect again --
+-- send again --
+{"kind": "request", "type": "hello", "id": "h1", "data": {"roomId": "$room", "ticket": "$ticketB", "protocolVersion": "$protocolVersion"}}
+-- receive again --
+{"kind": "response", "type": "hello", "id": "h1", "data": {}, "error": {"code": "UNAUTHORIZED", "message": "票据无效"}}
+-- closed again --
+4401
+
+-- request --
+GET /api/room
+Authorization: Bearer $ticketB
+-- response --
+401
+{"error": {"code": "UNAUTHORIZED", "message": "票据无效"}}
+
+-- request --
+POST /api/room/messages
+Authorization: Bearer $ticketA
+
+{"content": "还在"}
+-- response --
+204
+-- receive a --
+{"kind": "event", "type": "chat.message", "term": 1, "seq": 4, "data": {"member": {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}, "content": "还在", "sentAtMs": 1}}