diplomail (Stage A): add in-game personal mail subsystem

Phase 28 of ui/PLAN.md needs a persistent player-to-player mail channel; the existing `mail` package is a transactional email outbox and the `notification` catalog is one-way platform events. Stage A lands the schema (diplomail_messages / _recipients / _translations), a single-recipient personal send/read/delete service path, a `diplomail.message.received` push kind plumbed through the notification pipeline, and an unread-counts endpoint that drives the lobby badge. Admin / system mail, lifecycle hooks, paid-tier broadcast, multi-game broadcast, bulk purge and language detection / translation cache come in stages B–D. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-15 18:28:55 +02:00
parent 77cb7c78b6
commit 535e27008f
28 changed files with 3069 additions and 12 deletions
@@ -0,0 +1,473 @@
+package diplomail
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"time"
+
+	"galaxy/backend/internal/postgres/jet/backend/model"
+	"galaxy/backend/internal/postgres/jet/backend/table"
+
+	"github.com/go-jet/jet/v2/postgres"
+	"github.com/go-jet/jet/v2/qrm"
+	"github.com/google/uuid"
+)
+
+// Store is the Postgres-backed query surface for the diplomail
+// package. All queries are built through go-jet against the generated
+// table bindings under `backend/internal/postgres/jet/backend/table`.
+type Store struct {
+	db *sql.DB
+}
+
+// NewStore constructs a Store wrapping db.
+func NewStore(db *sql.DB) *Store { return &Store{db: db} }
+
+// messageColumns is the canonical projection for diplomail_messages
+// reads.
+func messageColumns() postgres.ColumnList {
+	m := table.DiplomailMessages
+	return postgres.ColumnList{
+		m.MessageID, m.GameID, m.GameName, m.Kind, m.SenderKind,
+		m.SenderUserID, m.SenderUsername, m.SenderIP,
+		m.Subject, m.Body, m.BodyLang, m.BroadcastScope, m.CreatedAt,
+	}
+}
+
+// recipientColumns is the canonical projection for
+// diplomail_recipients reads.
+func recipientColumns() postgres.ColumnList {
+	r := table.DiplomailRecipients
+	return postgres.ColumnList{
+		r.RecipientID, r.MessageID, r.GameID, r.UserID,
+		r.RecipientUserName, r.RecipientRaceName,
+		r.DeliveredAt, r.ReadAt, r.DeletedAt, r.NotifiedAt,
+	}
+}
+
+// MessageInsert carries the immutable per-message fields. The store
+// fills MessageID, sets CreatedAt to `now()` via the column default,
+// and leaves recipient-side state to InsertRecipient.
+type MessageInsert struct {
+	MessageID      uuid.UUID
+	GameID         uuid.UUID
+	GameName       string
+	Kind           string
+	SenderKind     string
+	SenderUserID   *uuid.UUID
+	SenderUsername *string
+	SenderIP       string
+	Subject        string
+	Body           string
+	BodyLang       string
+	BroadcastScope string
+}
+
+// RecipientInsert carries the per-recipient snapshot.
+type RecipientInsert struct {
+	RecipientID       uuid.UUID
+	MessageID         uuid.UUID
+	GameID            uuid.UUID
+	UserID            uuid.UUID
+	RecipientUserName string
+	RecipientRaceName *string
+}
+
+// InsertMessageWithRecipients persists a Message together with one or
+// more Recipient rows inside a single transaction. The function is
+// the canonical write path for every send variant: Stage A passes a
+// single-element slice; later stages reuse the same path for
+// broadcasts.
+func (s *Store) InsertMessageWithRecipients(ctx context.Context, msg MessageInsert, recipients []RecipientInsert) (Message, []Recipient, error) {
+	if len(recipients) == 0 {
+		return Message{}, nil, errors.New("diplomail store: at least one recipient required")
+	}
+
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return Message{}, nil, fmt.Errorf("diplomail store: begin tx: %w", err)
+	}
+	defer func() { _ = tx.Rollback() }()
+
+	m := table.DiplomailMessages
+	msgStmt := m.INSERT(
+		m.MessageID, m.GameID, m.GameName, m.Kind, m.SenderKind,
+		m.SenderUserID, m.SenderUsername, m.SenderIP,
+		m.Subject, m.Body, m.BodyLang, m.BroadcastScope,
+	).VALUES(
+		msg.MessageID,
+		msg.GameID,
+		msg.GameName,
+		msg.Kind,
+		msg.SenderKind,
+		uuidPtrArg(msg.SenderUserID),
+		stringPtrArg(msg.SenderUsername),
+		msg.SenderIP,
+		msg.Subject,
+		msg.Body,
+		msg.BodyLang,
+		msg.BroadcastScope,
+	).RETURNING(messageColumns())
+
+	var msgRow model.DiplomailMessages
+	if err := msgStmt.QueryContext(ctx, tx, &msgRow); err != nil {
+		return Message{}, nil, fmt.Errorf("diplomail store: insert message: %w", err)
+	}
+
+	r := table.DiplomailRecipients
+	rcptStmt := r.INSERT(
+		r.RecipientID, r.MessageID, r.GameID, r.UserID,
+		r.RecipientUserName, r.RecipientRaceName,
+	)
+	for _, in := range recipients {
+		rcptStmt = rcptStmt.VALUES(
+			in.RecipientID,
+			in.MessageID,
+			in.GameID,
+			in.UserID,
+			in.RecipientUserName,
+			stringPtrArg(in.RecipientRaceName),
+		)
+	}
+	rcptStmt = rcptStmt.RETURNING(recipientColumns())
+
+	var rcptRows []model.DiplomailRecipients
+	if err := rcptStmt.QueryContext(ctx, tx, &rcptRows); err != nil {
+		return Message{}, nil, fmt.Errorf("diplomail store: insert recipients: %w", err)
+	}
+
+	if err := tx.Commit(); err != nil {
+		return Message{}, nil, fmt.Errorf("diplomail store: commit: %w", err)
+	}
+
+	return messageFromModel(msgRow), recipientsFromModel(rcptRows), nil
+}
+
+// LoadMessage returns the Message row identified by messageID. The
+// function is used by readers that already verified recipient
+// authorisation; callers that need both the message and the
+// recipient's per-user state should use LoadInboxEntry.
+func (s *Store) LoadMessage(ctx context.Context, messageID uuid.UUID) (Message, error) {
+	m := table.DiplomailMessages
+	stmt := postgres.SELECT(messageColumns()).
+		FROM(m).
+		WHERE(m.MessageID.EQ(postgres.UUID(messageID))).
+		LIMIT(1)
+	var row model.DiplomailMessages
+	if err := stmt.QueryContext(ctx, s.db, &row); err != nil {
+		if errors.Is(err, qrm.ErrNoRows) {
+			return Message{}, ErrNotFound
+		}
+		return Message{}, fmt.Errorf("diplomail store: load message %s: %w", messageID, err)
+	}
+	return messageFromModel(row), nil
+}
+
+// LoadInboxEntry returns a Message together with the caller's
+// Recipient row, both for messageID. Returns ErrNotFound when the
+// caller is not a recipient of the message — this is also how the
+// service layer enforces "only recipients may read".
+func (s *Store) LoadInboxEntry(ctx context.Context, messageID, userID uuid.UUID) (InboxEntry, error) {
+	m := table.DiplomailMessages
+	r := table.DiplomailRecipients
+	cols := append(messageColumns(), recipientColumns()...)
+	stmt := postgres.SELECT(cols).
+		FROM(r.INNER_JOIN(m, m.MessageID.EQ(r.MessageID))).
+		WHERE(
+			r.MessageID.EQ(postgres.UUID(messageID)).
+				AND(r.UserID.EQ(postgres.UUID(userID))),
+		).
+		LIMIT(1)
+	var dest struct {
+		model.DiplomailMessages
+		Recipient model.DiplomailRecipients `alias:"diplomail_recipients"`
+	}
+	if err := stmt.QueryContext(ctx, s.db, &dest); err != nil {
+		if errors.Is(err, qrm.ErrNoRows) {
+			return InboxEntry{}, ErrNotFound
+		}
+		return InboxEntry{}, fmt.Errorf("diplomail store: load inbox entry %s/%s: %w", messageID, userID, err)
+	}
+	return InboxEntry{
+		Message:   messageFromModel(dest.DiplomailMessages),
+		Recipient: recipientFromModel(dest.Recipient),
+	}, nil
+}
+
+// ListInbox returns the recipient view of messages addressed to
+// userID in gameID, newest first. Soft-deleted rows
+// (`deleted_at IS NOT NULL`) are excluded.
+func (s *Store) ListInbox(ctx context.Context, gameID, userID uuid.UUID) ([]InboxEntry, error) {
+	m := table.DiplomailMessages
+	r := table.DiplomailRecipients
+	cols := append(messageColumns(), recipientColumns()...)
+	stmt := postgres.SELECT(cols).
+		FROM(r.INNER_JOIN(m, m.MessageID.EQ(r.MessageID))).
+		WHERE(
+			r.UserID.EQ(postgres.UUID(userID)).
+				AND(r.GameID.EQ(postgres.UUID(gameID))).
+				AND(r.DeletedAt.IS_NULL()),
+		).
+		ORDER_BY(m.CreatedAt.DESC(), m.MessageID.DESC())
+	var dest []struct {
+		model.DiplomailMessages
+		Recipient model.DiplomailRecipients `alias:"diplomail_recipients"`
+	}
+	if err := stmt.QueryContext(ctx, s.db, &dest); err != nil {
+		return nil, fmt.Errorf("diplomail store: list inbox %s/%s: %w", gameID, userID, err)
+	}
+	out := make([]InboxEntry, 0, len(dest))
+	for _, row := range dest {
+		out = append(out, InboxEntry{
+			Message:   messageFromModel(row.DiplomailMessages),
+			Recipient: recipientFromModel(row.Recipient),
+		})
+	}
+	return out, nil
+}
+
+// ListSent returns messages authored by senderUserID in gameID,
+// newest first. Personal messages only — admin/system rows have
+// `sender_user_id IS NULL` and are filtered out by the WHERE clause.
+func (s *Store) ListSent(ctx context.Context, gameID, senderUserID uuid.UUID) ([]Message, error) {
+	m := table.DiplomailMessages
+	stmt := postgres.SELECT(messageColumns()).
+		FROM(m).
+		WHERE(
+			m.GameID.EQ(postgres.UUID(gameID)).
+				AND(m.SenderUserID.EQ(postgres.UUID(senderUserID))),
+		).
+		ORDER_BY(m.CreatedAt.DESC(), m.MessageID.DESC())
+	var rows []model.DiplomailMessages
+	if err := stmt.QueryContext(ctx, s.db, &rows); err != nil {
+		return nil, fmt.Errorf("diplomail store: list sent %s/%s: %w", gameID, senderUserID, err)
+	}
+	out := make([]Message, 0, len(rows))
+	for _, row := range rows {
+		out = append(out, messageFromModel(row))
+	}
+	return out, nil
+}
+
+// MarkRead sets `read_at = at` on the recipient row identified by
+// (messageID, userID). Idempotent: a row that is already marked read
+// is left untouched but the existing Recipient is returned.
+// Returns ErrNotFound when the user is not a recipient of the message.
+func (s *Store) MarkRead(ctx context.Context, messageID, userID uuid.UUID, at time.Time) (Recipient, error) {
+	r := table.DiplomailRecipients
+	stmt := r.UPDATE(r.ReadAt).
+		SET(postgres.TimestampzT(at.UTC())).
+		WHERE(
+			r.MessageID.EQ(postgres.UUID(messageID)).
+				AND(r.UserID.EQ(postgres.UUID(userID))).
+				AND(r.ReadAt.IS_NULL()),
+		).
+		RETURNING(recipientColumns())
+	var row model.DiplomailRecipients
+	if err := stmt.QueryContext(ctx, s.db, &row); err != nil {
+		if !errors.Is(err, qrm.ErrNoRows) {
+			return Recipient{}, fmt.Errorf("diplomail store: mark read %s/%s: %w", messageID, userID, err)
+		}
+		// The row exists but read_at was already set, or the row
+		// does not exist at all. Fetch to disambiguate.
+		existing, loadErr := s.LoadRecipient(ctx, messageID, userID)
+		if loadErr != nil {
+			return Recipient{}, loadErr
+		}
+		return existing, nil
+	}
+	return recipientFromModel(row), nil
+}
+
+// SoftDelete sets `deleted_at = at` on the recipient row identified by
+// (messageID, userID). The row must already have `read_at` set;
+// otherwise the call returns ErrConflict so a hostile client cannot
+// erase a message before opening it (item 10 of the spec).
+// Returns ErrNotFound when the user is not a recipient.
+func (s *Store) SoftDelete(ctx context.Context, messageID, userID uuid.UUID, at time.Time) (Recipient, error) {
+	r := table.DiplomailRecipients
+	stmt := r.UPDATE(r.DeletedAt).
+		SET(postgres.TimestampzT(at.UTC())).
+		WHERE(
+			r.MessageID.EQ(postgres.UUID(messageID)).
+				AND(r.UserID.EQ(postgres.UUID(userID))).
+				AND(r.ReadAt.IS_NOT_NULL()).
+				AND(r.DeletedAt.IS_NULL()),
+		).
+		RETURNING(recipientColumns())
+	var row model.DiplomailRecipients
+	if err := stmt.QueryContext(ctx, s.db, &row); err != nil {
+		if !errors.Is(err, qrm.ErrNoRows) {
+			return Recipient{}, fmt.Errorf("diplomail store: soft delete %s/%s: %w", messageID, userID, err)
+		}
+		existing, loadErr := s.LoadRecipient(ctx, messageID, userID)
+		if loadErr != nil {
+			return Recipient{}, loadErr
+		}
+		if existing.ReadAt == nil {
+			return Recipient{}, fmt.Errorf("%w: message must be read before delete", ErrConflict)
+		}
+		// Already deleted: return the existing row idempotently.
+		return existing, nil
+	}
+	return recipientFromModel(row), nil
+}
+
+// LoadRecipient fetches the Recipient row keyed on (messageID, userID).
+// Returns ErrNotFound when no such recipient exists.
+func (s *Store) LoadRecipient(ctx context.Context, messageID, userID uuid.UUID) (Recipient, error) {
+	r := table.DiplomailRecipients
+	stmt := postgres.SELECT(recipientColumns()).
+		FROM(r).
+		WHERE(
+			r.MessageID.EQ(postgres.UUID(messageID)).
+				AND(r.UserID.EQ(postgres.UUID(userID))),
+		).
+		LIMIT(1)
+	var row model.DiplomailRecipients
+	if err := stmt.QueryContext(ctx, s.db, &row); err != nil {
+		if errors.Is(err, qrm.ErrNoRows) {
+			return Recipient{}, ErrNotFound
+		}
+		return Recipient{}, fmt.Errorf("diplomail store: load recipient %s/%s: %w", messageID, userID, err)
+	}
+	return recipientFromModel(row), nil
+}
+
+// UnreadCountForUserGame returns the count of unread, non-deleted
+// messages addressed to userID in gameID. Backs the push payload
+// `unread_game` field.
+func (s *Store) UnreadCountForUserGame(ctx context.Context, gameID, userID uuid.UUID) (int, error) {
+	r := table.DiplomailRecipients
+	stmt := postgres.SELECT(postgres.COUNT(postgres.STAR).AS("count")).
+		FROM(r).
+		WHERE(
+			r.UserID.EQ(postgres.UUID(userID)).
+				AND(r.GameID.EQ(postgres.UUID(gameID))).
+				AND(r.ReadAt.IS_NULL()).
+				AND(r.DeletedAt.IS_NULL()),
+		)
+	var dest struct {
+		Count int64 `alias:"count"`
+	}
+	if err := stmt.QueryContext(ctx, s.db, &dest); err != nil {
+		return 0, fmt.Errorf("diplomail store: unread count %s/%s: %w", gameID, userID, err)
+	}
+	return int(dest.Count), nil
+}
+
+// UnreadCountsForUser returns a per-game breakdown of unread messages
+// addressed to userID, plus the matching game names so the lobby
+// badge UI can render entries even after the recipient's membership
+// has been revoked. The slice is ordered by game name.
+func (s *Store) UnreadCountsForUser(ctx context.Context, userID uuid.UUID) ([]UnreadCount, error) {
+	r := table.DiplomailRecipients
+	m := table.DiplomailMessages
+	stmt := postgres.SELECT(
+		r.GameID.AS("game_id"),
+		postgres.MAX(m.GameName).AS("game_name"),
+		postgres.COUNT(postgres.STAR).AS("count"),
+	).
+		FROM(r.INNER_JOIN(m, m.MessageID.EQ(r.MessageID))).
+		WHERE(
+			r.UserID.EQ(postgres.UUID(userID)).
+				AND(r.ReadAt.IS_NULL()).
+				AND(r.DeletedAt.IS_NULL()),
+		).
+		GROUP_BY(r.GameID).
+		ORDER_BY(postgres.MAX(m.GameName).ASC())
+	var dest []struct {
+		GameID   uuid.UUID `alias:"game_id"`
+		GameName string    `alias:"game_name"`
+		Count    int64     `alias:"count"`
+	}
+	if err := stmt.QueryContext(ctx, s.db, &dest); err != nil {
+		return nil, fmt.Errorf("diplomail store: unread counts %s: %w", userID, err)
+	}
+	out := make([]UnreadCount, 0, len(dest))
+	for _, row := range dest {
+		out = append(out, UnreadCount{
+			GameID:   row.GameID,
+			GameName: row.GameName,
+			Unread:   int(row.Count),
+		})
+	}
+	return out, nil
+}
+
+// messageFromModel converts a jet-generated row to the domain type.
+func messageFromModel(row model.DiplomailMessages) Message {
+	out := Message{
+		MessageID:      row.MessageID,
+		GameID:         row.GameID,
+		GameName:       row.GameName,
+		Kind:           row.Kind,
+		SenderKind:     row.SenderKind,
+		SenderIP:       row.SenderIP,
+		Subject:        row.Subject,
+		Body:           row.Body,
+		BodyLang:       row.BodyLang,
+		BroadcastScope: row.BroadcastScope,
+		CreatedAt:      row.CreatedAt,
+	}
+	if row.SenderUserID != nil {
+		id := *row.SenderUserID
+		out.SenderUserID = &id
+	}
+	if row.SenderUsername != nil {
+		name := *row.SenderUsername
+		out.SenderUsername = &name
+	}
+	return out
+}
+
+// recipientFromModel converts a jet-generated row to the domain type.
+func recipientFromModel(row model.DiplomailRecipients) Recipient {
+	out := Recipient{
+		RecipientID:       row.RecipientID,
+		MessageID:         row.MessageID,
+		GameID:            row.GameID,
+		UserID:            row.UserID,
+		RecipientUserName: row.RecipientUserName,
+		DeliveredAt:       row.DeliveredAt,
+		ReadAt:            row.ReadAt,
+		DeletedAt:         row.DeletedAt,
+		NotifiedAt:        row.NotifiedAt,
+	}
+	if row.RecipientRaceName != nil {
+		name := *row.RecipientRaceName
+		out.RecipientRaceName = &name
+	}
+	return out
+}
+
+// recipientsFromModel converts a slice in place. Used by
+// InsertMessageWithRecipients.
+func recipientsFromModel(rows []model.DiplomailRecipients) []Recipient {
+	out := make([]Recipient, 0, len(rows))
+	for _, row := range rows {
+		out = append(out, recipientFromModel(row))
+	}
+	return out
+}
+
+// uuidPtrArg returns the jet argument expression for a nullable UUID.
+// Pre-NULL handling here avoids a custom NULL literal at every call
+// site.
+func uuidPtrArg(v *uuid.UUID) postgres.Expression {
+	if v == nil {
+		return postgres.NULL
+	}
+	return postgres.UUID(*v)
+}
+
+// stringPtrArg returns the jet argument expression for a nullable
+// text column.
+func stringPtrArg(v *string) postgres.Expression {
+	if v == nil {
+		return postgres.NULL
+	}
+	return postgres.String(*v)
+}