Stage 11: account linking & merge (email + Telegram Login Widget) (#12)

2026-06-04 09:18:17 +00:00
parent 3a640a17a4
commit 01485d8fc6
68 changed files with 3331 additions and 369 deletions
@@ -0,0 +1,497 @@
+// Package accountmerge retires a secondary account into a primary one in a single
+// transaction: it sums statistics and the hint wallet, ORs the paid flag, repoints
+// the secondary's identities, transfers its games/chat/complaints/invitations,
+// de-duplicates friends and blocks, and leaves the secondary as an audit tombstone
+// (accounts.merged_into). It is the data core of Stage 11 account linking & merge
+// (ARCHITECTURE.md §4); session revocation and any session switch are orchestrated
+// one layer up (the link service), since the in-memory session cache lives there.
+package accountmerge
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/go-jet/jet/v2/postgres"
+	"github.com/go-jet/jet/v2/qrm"
+	"github.com/google/uuid"
+
+	"scrabble/backend/internal/postgres/jet/backend/model"
+	"scrabble/backend/internal/postgres/jet/backend/table"
+)
+
+// statusActive mirrors game.StatusActive; the active-shared-game guard reads it
+// without taking a dependency on the game package.
+const statusActive = "active"
+
+// Friendship statuses, highest precedence first, mirroring internal/social.
+const (
+	friendAccepted = "accepted"
+	friendPending  = "pending"
+	friendDeclined = "declined"
+)
+
+// ErrActiveGameConflict is returned when the primary and secondary accounts share
+// an active game: merging would seat one player against themselves, so the caller
+// must wait for the game to finish.
+var ErrActiveGameConflict = errors.New("accountmerge: primary and secondary share an active game")
+
+// ErrSameAccount is returned when primary and secondary are the same account.
+var ErrSameAccount = errors.New("accountmerge: primary and secondary are the same account")
+
+// Merger merges accounts over a Postgres handle.
+type Merger struct {
+	db  *sql.DB
+	now func() time.Time
+}
+
+// NewMerger constructs a Merger over db.
+func NewMerger(db *sql.DB) *Merger {
+	return &Merger{db: db, now: func() time.Time { return time.Now().UTC() }}
+}
+
+// Merge retires secondary into primary atomically. The secondary is kept as a
+// tombstone (merged_into=primary) so the no-cascade foreign keys of any shared
+// finished game stay valid; its seat in such a game is left untouched. The merge
+// is refused with ErrActiveGameConflict when the two share an active game.
+func (m *Merger) Merge(ctx context.Context, primary, secondary uuid.UUID) error {
+	if primary == secondary {
+		return ErrSameAccount
+	}
+	now := m.now()
+	return withTx(ctx, m.db, func(tx *sql.Tx) error {
+		if err := guardActiveSharedGame(ctx, tx, primary, secondary); err != nil {
+			return err
+		}
+		if err := mergeStats(ctx, tx, primary, secondary, now); err != nil {
+			return err
+		}
+		if err := mergeAccountFields(ctx, tx, primary, secondary, now); err != nil {
+			return err
+		}
+		if err := reassignColumn(ctx, tx, table.Identities, table.Identities.AccountID, primary, secondary); err != nil {
+			return fmt.Errorf("accountmerge: identities: %w", err)
+		}
+		if err := transferGamePlayers(ctx, tx, primary, secondary); err != nil {
+			return err
+		}
+		if err := reassignColumn(ctx, tx, table.ChatMessages, table.ChatMessages.SenderID, primary, secondary); err != nil {
+			return fmt.Errorf("accountmerge: chat: %w", err)
+		}
+		if err := reassignColumn(ctx, tx, table.Complaints, table.Complaints.ComplainantID, primary, secondary); err != nil {
+			return fmt.Errorf("accountmerge: complaints: %w", err)
+		}
+		if err := mergeFriendships(ctx, tx, primary, secondary); err != nil {
+			return err
+		}
+		if err := mergeBlocks(ctx, tx, primary, secondary); err != nil {
+			return err
+		}
+		if err := mergeInvitations(ctx, tx, primary, secondary); err != nil {
+			return err
+		}
+		if err := deleteEphemerals(ctx, tx, secondary); err != nil {
+			return err
+		}
+		return tombstone(ctx, tx, primary, secondary, now)
+	})
+}
+
+// guardActiveSharedGame returns ErrActiveGameConflict when primary and secondary
+// are both seated in the same active game.
+func guardActiveSharedGame(ctx context.Context, tx *sql.Tx, primary, secondary uuid.UUID) error {
+	pri, err := activeGameIDs(ctx, tx, primary)
+	if err != nil {
+		return err
+	}
+	if len(pri) == 0 {
+		return nil
+	}
+	sec, err := activeGameIDs(ctx, tx, secondary)
+	if err != nil {
+		return err
+	}
+	have := make(map[uuid.UUID]struct{}, len(pri))
+	for _, id := range pri {
+		have[id] = struct{}{}
+	}
+	for _, id := range sec {
+		if _, ok := have[id]; ok {
+			return ErrActiveGameConflict
+		}
+	}
+	return nil
+}
+
+// activeGameIDs lists the active games accountID is seated in.
+func activeGameIDs(ctx context.Context, tx *sql.Tx, accountID uuid.UUID) ([]uuid.UUID, error) {
+	stmt := postgres.SELECT(table.GamePlayers.GameID).
+		FROM(table.GamePlayers.INNER_JOIN(table.Games, table.Games.GameID.EQ(table.GamePlayers.GameID))).
+		WHERE(
+			table.GamePlayers.AccountID.EQ(postgres.UUID(accountID)).
+				AND(table.Games.Status.EQ(postgres.String(statusActive))),
+		)
+	var rows []model.GamePlayers
+	if err := stmt.QueryContext(ctx, tx, &rows); err != nil {
+		if errors.Is(err, qrm.ErrNoRows) {
+			return nil, nil
+		}
+		return nil, fmt.Errorf("accountmerge: active games %s: %w", accountID, err)
+	}
+	out := make([]uuid.UUID, 0, len(rows))
+	for _, r := range rows {
+		out = append(out, r.GameID)
+	}
+	return out, nil
+}
+
+// mergeStats folds secondary's lifetime statistics into primary (wins/losses/draws
+// summed, max points kept) and deletes the secondary row.
+func mergeStats(ctx context.Context, tx *sql.Tx, primary, secondary uuid.UUID, now time.Time) error {
+	var sec model.AccountStats
+	err := postgres.SELECT(table.AccountStats.AllColumns).
+		FROM(table.AccountStats).
+		WHERE(table.AccountStats.AccountID.EQ(postgres.UUID(secondary))).
+		QueryContext(ctx, tx, &sec)
+	if errors.Is(err, qrm.ErrNoRows) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("accountmerge: load secondary stats: %w", err)
+	}
+
+	ensure := table.AccountStats.INSERT(table.AccountStats.AccountID).
+		VALUES(primary).ON_CONFLICT(table.AccountStats.AccountID).DO_NOTHING()
+	if _, err := ensure.ExecContext(ctx, tx); err != nil {
+		return fmt.Errorf("accountmerge: ensure primary stats: %w", err)
+	}
+	var pri model.AccountStats
+	if err := postgres.SELECT(table.AccountStats.AllColumns).
+		FROM(table.AccountStats).
+		WHERE(table.AccountStats.AccountID.EQ(postgres.UUID(primary))).
+		FOR(postgres.UPDATE()).
+		QueryContext(ctx, tx, &pri); err != nil {
+		return fmt.Errorf("accountmerge: lock primary stats: %w", err)
+	}
+
+	upd := table.AccountStats.UPDATE(
+		table.AccountStats.Wins, table.AccountStats.Losses, table.AccountStats.Draws,
+		table.AccountStats.MaxGamePoints, table.AccountStats.MaxWordPoints, table.AccountStats.UpdatedAt,
+	).SET(
+		postgres.Int(int64(pri.Wins+sec.Wins)),
+		postgres.Int(int64(pri.Losses+sec.Losses)),
+		postgres.Int(int64(pri.Draws+sec.Draws)),
+		postgres.Int(int64(max(pri.MaxGamePoints, sec.MaxGamePoints))),
+		postgres.Int(int64(max(pri.MaxWordPoints, sec.MaxWordPoints))),
+		postgres.TimestampzT(now),
+	).WHERE(table.AccountStats.AccountID.EQ(postgres.UUID(primary)))
+	if _, err := upd.ExecContext(ctx, tx); err != nil {
+		return fmt.Errorf("accountmerge: update primary stats: %w", err)
+	}
+
+	del := table.AccountStats.DELETE().WHERE(table.AccountStats.AccountID.EQ(postgres.UUID(secondary)))
+	if _, err := del.ExecContext(ctx, tx); err != nil {
+		return fmt.Errorf("accountmerge: delete secondary stats: %w", err)
+	}
+	return nil
+}
+
+// mergeAccountFields adds secondary's hint wallet to primary and ORs the paid flag;
+// all other profile fields stay the primary's.
+func mergeAccountFields(ctx context.Context, tx *sql.Tx, primary, secondary uuid.UUID, now time.Time) error {
+	var sec model.Accounts
+	if err := postgres.SELECT(table.Accounts.AllColumns).
+		FROM(table.Accounts).
+		WHERE(table.Accounts.AccountID.EQ(postgres.UUID(secondary))).
+		QueryContext(ctx, tx, &sec); err != nil {
+		return fmt.Errorf("accountmerge: load secondary account: %w", err)
+	}
+	upd := table.Accounts.UPDATE(
+		table.Accounts.HintBalance, table.Accounts.PaidAccount, table.Accounts.UpdatedAt,
+	).SET(
+		table.Accounts.HintBalance.ADD(postgres.Int(int64(sec.HintBalance))),
+		table.Accounts.PaidAccount.OR(postgres.Bool(sec.PaidAccount)),
+		postgres.TimestampzT(now),
+	).WHERE(table.Accounts.AccountID.EQ(postgres.UUID(primary)))
+	if _, err := upd.ExecContext(ctx, tx); err != nil {
+		return fmt.Errorf("accountmerge: update primary account: %w", err)
+	}
+	return nil
+}
+
+// transferGamePlayers moves secondary's seats to primary, except in a game primary
+// already sits in (a shared finished game — active is barred by the guard), where
+// the secondary seat is left as the tombstone so the no-cascade FK stays valid.
+func transferGamePlayers(ctx context.Context, tx *sql.Tx, primary, secondary uuid.UUID) error {
+	var prows []model.GamePlayers
+	if err := postgres.SELECT(table.GamePlayers.GameID).
+		FROM(table.GamePlayers).
+		WHERE(table.GamePlayers.AccountID.EQ(postgres.UUID(primary))).
+		QueryContext(ctx, tx, &prows); err != nil {
+		if !errors.Is(err, qrm.ErrNoRows) {
+			return fmt.Errorf("accountmerge: primary seats: %w", err)
+		}
+	}
+	cond := table.GamePlayers.AccountID.EQ(postgres.UUID(secondary))
+	if len(prows) > 0 {
+		ids := make([]postgres.Expression, len(prows))
+		for i, r := range prows {
+			ids[i] = postgres.UUID(r.GameID)
+		}
+		cond = cond.AND(table.GamePlayers.GameID.NOT_IN(ids...))
+	}
+	upd := table.GamePlayers.UPDATE(table.GamePlayers.AccountID).SET(postgres.UUID(primary)).WHERE(cond)
+	if _, err := upd.ExecContext(ctx, tx); err != nil {
+		return fmt.Errorf("accountmerge: transfer seats: %w", err)
+	}
+	return nil
+}
+
+// reassignColumn blanket-reassigns a no-collision account column from secondary to
+// primary (identities, chat sender, complaint complainant).
+func reassignColumn(ctx context.Context, tx *sql.Tx, tbl postgres.Table, col postgres.ColumnString, primary, secondary uuid.UUID) error {
+	upd := tbl.UPDATE(col).SET(postgres.UUID(primary)).
+		WHERE(col.EQ(postgres.UUID(secondary)))
+	_, err := upd.ExecContext(ctx, tx)
+	return err
+}
+
+// friendRank ranks a friendship status for dedupe precedence (higher wins).
+func friendRank(status string) int {
+	switch status {
+	case friendAccepted:
+		return 3
+	case friendPending:
+		return 2
+	case friendDeclined:
+		return 1
+	default:
+		return 0
+	}
+}
+
+// mergeFriendships repoints secondary's friendships to primary, dropping the direct
+// primary-secondary edge (it would become a self-edge) and de-duplicating a shared
+// counterparty by keeping the higher-precedence status (accepted > pending >
+// declined). Each account has at most one edge per unordered pair, so the per-other
+// decision is unambiguous.
+func mergeFriendships(ctx context.Context, tx *sql.Tx, primary, secondary uuid.UUID) error {
+	if err := deletePair(ctx, tx, table.Friendships.DELETE(),
+		table.Friendships.RequesterID, table.Friendships.AddresseeID, primary, secondary); err != nil {
+		return fmt.Errorf("accountmerge: drop self-friendship: %w", err)
+	}
+
+	priByOther := map[uuid.UUID]string{}
+	var prows []model.Friendships
+	if err := selectEdges(ctx, tx, table.Friendships, table.Friendships.AllColumns, table.Friendships.RequesterID, table.Friendships.AddresseeID, primary, &prows); err != nil {
+		return fmt.Errorf("accountmerge: primary friendships: %w", err)
+	}
+	for _, r := range prows {
+		priByOther[otherOf(r.RequesterID, r.AddresseeID, primary)] = r.Status
+	}
+
+	var srows []model.Friendships
+	if err := selectEdges(ctx, tx, table.Friendships, table.Friendships.AllColumns, table.Friendships.RequesterID, table.Friendships.AddresseeID, secondary, &srows); err != nil {
+		return fmt.Errorf("accountmerge: secondary friendships: %w", err)
+	}
+	for _, r := range srows {
+		other := otherOf(r.RequesterID, r.AddresseeID, secondary)
+		if priStatus, ok := priByOther[other]; ok {
+			if friendRank(r.Status) <= friendRank(priStatus) {
+				if err := deleteEdge(ctx, tx, table.Friendships.DELETE(),
+					table.Friendships.RequesterID, table.Friendships.AddresseeID, r.RequesterID, r.AddresseeID); err != nil {
+					return fmt.Errorf("accountmerge: drop dominated friendship: %w", err)
+				}
+				continue
+			}
+			if err := deletePair(ctx, tx, table.Friendships.DELETE(),
+				table.Friendships.RequesterID, table.Friendships.AddresseeID, primary, other); err != nil {
+				return fmt.Errorf("accountmerge: drop superseded friendship: %w", err)
+			}
+		}
+		if err := repointEdge(ctx, tx, table.Friendships, table.Friendships.RequesterID, table.Friendships.AddresseeID,
+			r.RequesterID, r.AddresseeID, primary, secondary); err != nil {
+			return fmt.Errorf("accountmerge: repoint friendship: %w", err)
+		}
+	}
+	return nil
+}
+
+// mergeBlocks repoints secondary's blocks to primary, dropping the direct
+// primary-secondary block (a self-block) and de-duplicating a counterparty already
+// blocked by primary in either direction (a block is undirected for suppression).
+func mergeBlocks(ctx context.Context, tx *sql.Tx, primary, secondary uuid.UUID) error {
+	if err := deletePair(ctx, tx, table.Blocks.DELETE(),
+		table.Blocks.BlockerID, table.Blocks.BlockedID, primary, secondary); err != nil {
+		return fmt.Errorf("accountmerge: drop self-block: %w", err)
+	}
+
+	priOthers := map[uuid.UUID]struct{}{}
+	var prows []model.Blocks
+	if err := selectEdges(ctx, tx, table.Blocks, table.Blocks.AllColumns, table.Blocks.BlockerID, table.Blocks.BlockedID, primary, &prows); err != nil {
+		return fmt.Errorf("accountmerge: primary blocks: %w", err)
+	}
+	for _, r := range prows {
+		priOthers[otherOf(r.BlockerID, r.BlockedID, primary)] = struct{}{}
+	}
+
+	var srows []model.Blocks
+	if err := selectEdges(ctx, tx, table.Blocks, table.Blocks.AllColumns, table.Blocks.BlockerID, table.Blocks.BlockedID, secondary, &srows); err != nil {
+		return fmt.Errorf("accountmerge: secondary blocks: %w", err)
+	}
+	for _, r := range srows {
+		if _, ok := priOthers[otherOf(r.BlockerID, r.BlockedID, secondary)]; ok {
+			if err := deleteEdge(ctx, tx, table.Blocks.DELETE(),
+				table.Blocks.BlockerID, table.Blocks.BlockedID, r.BlockerID, r.BlockedID); err != nil {
+				return fmt.Errorf("accountmerge: drop dup block: %w", err)
+			}
+			continue
+		}
+		if err := repointEdge(ctx, tx, table.Blocks, table.Blocks.BlockerID, table.Blocks.BlockedID,
+			r.BlockerID, r.BlockedID, primary, secondary); err != nil {
+			return fmt.Errorf("accountmerge: repoint block: %w", err)
+		}
+	}
+	return nil
+}
+
+// mergeInvitations deletes secondary's pending invitations as inviter (cascading to
+// their invitees) and repoints its invitee rows to primary, dropping a row where
+// primary is already an invitee of the same invitation.
+func mergeInvitations(ctx context.Context, tx *sql.Tx, primary, secondary uuid.UUID) error {
+	delInv := table.GameInvitations.DELETE().
+		WHERE(table.GameInvitations.InviterID.EQ(postgres.UUID(secondary)))
+	if _, err := delInv.ExecContext(ctx, tx); err != nil {
+		return fmt.Errorf("accountmerge: delete secondary invitations: %w", err)
+	}
+
+	priInv := map[uuid.UUID]struct{}{}
+	var prows []model.GameInvitationInvitees
+	if err := postgres.SELECT(table.GameInvitationInvitees.InvitationID).
+		FROM(table.GameInvitationInvitees).
+		WHERE(table.GameInvitationInvitees.AccountID.EQ(postgres.UUID(primary))).
+		QueryContext(ctx, tx, &prows); err != nil && !errors.Is(err, qrm.ErrNoRows) {
+		return fmt.Errorf("accountmerge: primary invitees: %w", err)
+	}
+	for _, r := range prows {
+		priInv[r.InvitationID] = struct{}{}
+	}
+
+	var srows []model.GameInvitationInvitees
+	if err := postgres.SELECT(table.GameInvitationInvitees.InvitationID).
+		FROM(table.GameInvitationInvitees).
+		WHERE(table.GameInvitationInvitees.AccountID.EQ(postgres.UUID(secondary))).
+		QueryContext(ctx, tx, &srows); err != nil && !errors.Is(err, qrm.ErrNoRows) {
+		return fmt.Errorf("accountmerge: secondary invitees: %w", err)
+	}
+	for _, r := range srows {
+		where := table.GameInvitationInvitees.InvitationID.EQ(postgres.UUID(r.InvitationID)).
+			AND(table.GameInvitationInvitees.AccountID.EQ(postgres.UUID(secondary)))
+		if _, dup := priInv[r.InvitationID]; dup {
+			if _, err := table.GameInvitationInvitees.DELETE().WHERE(where).ExecContext(ctx, tx); err != nil {
+				return fmt.Errorf("accountmerge: drop dup invitee: %w", err)
+			}
+			continue
+		}
+		upd := table.GameInvitationInvitees.UPDATE(table.GameInvitationInvitees.AccountID).
+			SET(postgres.UUID(primary)).WHERE(where)
+		if _, err := upd.ExecContext(ctx, tx); err != nil {
+			return fmt.Errorf("accountmerge: repoint invitee: %w", err)
+		}
+	}
+	return nil
+}
+
+// deleteEphemerals drops the secondary's pending email confirmations and friend
+// codes (short-lived, single-use; not worth carrying over).
+func deleteEphemerals(ctx context.Context, tx *sql.Tx, secondary uuid.UUID) error {
+	if _, err := table.EmailConfirmations.DELETE().
+		WHERE(table.EmailConfirmations.AccountID.EQ(postgres.UUID(secondary))).
+		ExecContext(ctx, tx); err != nil {
+		return fmt.Errorf("accountmerge: delete confirmations: %w", err)
+	}
+	if _, err := table.FriendCodes.DELETE().
+		WHERE(table.FriendCodes.AccountID.EQ(postgres.UUID(secondary))).
+		ExecContext(ctx, tx); err != nil {
+		return fmt.Errorf("accountmerge: delete friend codes: %w", err)
+	}
+	return nil
+}
+
+// tombstone marks secondary retired, pointing at primary for audit.
+func tombstone(ctx context.Context, tx *sql.Tx, primary, secondary uuid.UUID, now time.Time) error {
+	upd := table.Accounts.UPDATE(table.Accounts.MergedInto, table.Accounts.MergedAt, table.Accounts.UpdatedAt).
+		SET(postgres.UUID(primary), postgres.TimestampzT(now), postgres.TimestampzT(now)).
+		WHERE(table.Accounts.AccountID.EQ(postgres.UUID(secondary)))
+	if _, err := upd.ExecContext(ctx, tx); err != nil {
+		return fmt.Errorf("accountmerge: tombstone secondary: %w", err)
+	}
+	return nil
+}
+
+// otherOf returns the endpoint of a two-account edge that is not self.
+func otherOf(a, b, self uuid.UUID) uuid.UUID {
+	if a == self {
+		return b
+	}
+	return a
+}
+
+// selectEdges loads the rows of a symmetric two-column edge table touching account.
+func selectEdges[T any](ctx context.Context, tx *sql.Tx, tbl postgres.Table, cols postgres.Projection, left, right postgres.ColumnString, account uuid.UUID, dest *[]T) error {
+	err := postgres.SELECT(cols).
+		FROM(tbl).
+		WHERE(left.EQ(postgres.UUID(account)).OR(right.EQ(postgres.UUID(account)))).
+		QueryContext(ctx, tx, dest)
+	if errors.Is(err, qrm.ErrNoRows) {
+		return nil
+	}
+	return err
+}
+
+// deletePair deletes the directed-or-reverse edge between a and b.
+func deletePair(ctx context.Context, tx *sql.Tx, del postgres.DeleteStatement, left, right postgres.ColumnString, a, b uuid.UUID) error {
+	cond := left.EQ(postgres.UUID(a)).AND(right.EQ(postgres.UUID(b))).
+		OR(left.EQ(postgres.UUID(b)).AND(right.EQ(postgres.UUID(a))))
+	_, err := del.WHERE(cond).ExecContext(ctx, tx)
+	return err
+}
+
+// deleteEdge deletes the single edge identified by its (left, right) primary key.
+func deleteEdge(ctx context.Context, tx *sql.Tx, del postgres.DeleteStatement, left, right postgres.ColumnString, l, r uuid.UUID) error {
+	cond := left.EQ(postgres.UUID(l)).AND(right.EQ(postgres.UUID(r)))
+	_, err := del.WHERE(cond).ExecContext(ctx, tx)
+	return err
+}
+
+// repointEdge replaces the secondary endpoint of edge (l, r) with primary, keeping
+// the edge's direction.
+func repointEdge(ctx context.Context, tx *sql.Tx, tbl postgres.Table, left, right postgres.ColumnString, l, r, primary, secondary uuid.UUID) error {
+	var col postgres.ColumnString
+	var where postgres.BoolExpression
+	if l == secondary {
+		col, where = left, left.EQ(postgres.UUID(secondary)).AND(right.EQ(postgres.UUID(r)))
+	} else {
+		col, where = right, left.EQ(postgres.UUID(l)).AND(right.EQ(postgres.UUID(secondary)))
+	}
+	_, err := tbl.UPDATE(col).SET(postgres.UUID(primary)).WHERE(where).ExecContext(ctx, tx)
+	return err
+}
+
+// withTx wraps fn in a transaction, committing on nil and rolling back on error.
+func withTx(ctx context.Context, db *sql.DB, fn func(tx *sql.Tx) error) error {
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("accountmerge: begin tx: %w", err)
+	}
+	if err := fn(tx); err != nil {
+		_ = tx.Rollback()
+		return err
+	}
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("accountmerge: commit tx: %w", err)
+	}
+	return nil
+}