feat(social): add friends, private chats, inbox and music sharing

services/api/internal/auth/middleware.go

@@ -122,6 +122,9 @@ func Middleware(svc *Service, log *slog.Logger) func(http.Handler) http.Handler
 			if route == "POST /api/v1/admin/users/import" {
 				r.Body = http.MaxBytesReader(w, r.Body, MaxUserArchiveBytes)
 			}
+			if r.Method == http.MethodPost && strings.HasPrefix(r.URL.Path, "/api/v1/me/social/") {
+				r.Body = http.MaxBytesReader(w, r.Body, 32<<10)
+			}
 			next.ServeHTTP(w, r.WithContext(WithPrincipal(ctx, p)))
 		})
 	}

services/api/internal/server/router.go

@@ -17,6 +17,7 @@ import (
 	"pipu/api/internal/platform/httpx"
 	"pipu/api/internal/playback"
 	"pipu/api/internal/search"
+	"pipu/api/internal/social"
 	"pipu/api/internal/system"
 	"pipu/api/internal/uploads"
 	"pipu/api/internal/wave"
@@ -24,6 +25,7 @@ import (
 
 // Deps — модули, реализующие части контракта API.
 type Deps struct {
+	Social     *social.Handler
 	Management *management.Handler
 	Genius     *genius.Handler
 	Log        *slog.Logger
@@ -48,6 +50,7 @@ type Deps struct {
 // По мере появления модулей (catalog, ...) они встраиваются сюда.
 // Алиасы дают встроенным полям разные имена: у модулей одинаковое имя типа Handler.
 type (
+	socialHandler     = social.Handler
 	managementHandler = management.Handler
 	geniusHandler     = genius.Handler
 	systemHandler     = system.Handler
@@ -61,6 +64,7 @@ type (
 )
 
 type handlers struct {
+	*socialHandler
 	*managementHandler
 	*geniusHandler
 	*systemHandler
@@ -93,7 +97,7 @@ func NewRouter(d Deps) http.Handler {
 		d.TUS.Mount(r)
 	}
 
-	strict := gen.NewStrictHandlerWithOptions(handlers{managementHandler: d.Management, geniusHandler: d.Genius, systemHandler: d.System, authHandler: d.Auth, libraryHandler: d.Library, searchHandler: d.Search, playbackHandler: d.Playback, waveHandler: d.Wave, appHandler: d.App, uploadsHandler: d.Uploads},
+	strict := gen.NewStrictHandlerWithOptions(handlers{socialHandler: d.Social, managementHandler: d.Management, geniusHandler: d.Genius, systemHandler: d.System, authHandler: d.Auth, libraryHandler: d.Library, searchHandler: d.Search, playbackHandler: d.Playback, waveHandler: d.Wave, appHandler: d.App, uploadsHandler: d.Uploads},
 		[]gen.StrictMiddlewareFunc{httpx.StrictRequest}, gen.StrictHTTPServerOptions{
 			RequestErrorHandlerFunc:  httpx.RequestError,
 			ResponseErrorHandlerFunc: httpx.ResponseError(log),

services/api/internal/social/handler.go

@@ -0,0 +1,108 @@
+// Package social owns friendships, private conversations and the persistent inbox.
+package social
+
+import (
+	"context"
+	"strings"
+	"unicode/utf8"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"pipu/api/internal/api/gen"
+	"pipu/api/internal/auth"
+	"pipu/api/internal/platform/apperr"
+)
+
+type Handler struct{ store *Store }
+
+func NewHandler(pool *pgxpool.Pool) *Handler { return &Handler{store: &Store{pool: pool}} }
+func userID(ctx context.Context) (uuid.UUID, error) {
+	p, ok := auth.PrincipalFrom(ctx)
+	if !ok {
+		return uuid.Nil, apperr.Unauthorized("auth.unauthorized", "sign in required")
+	}
+	return p.UserID, nil
+}
+func invalid() error {
+	return apperr.Validation("social", "social.invalid", "invalid social operation", nil)
+}
+func denied() error {
+	return apperr.Forbidden("social.unavailable", "friendship required or person unavailable")
+}
+
+func (h *Handler) GetSocial(ctx context.Context, _ gen.GetSocialRequestObject) (gen.GetSocialResponseObject, error) {
+	id, e := userID(ctx)
+	if e != nil {
+		return nil, e
+	}
+	v, e := h.store.inbox(ctx, id)
+	return gen.GetSocial200JSONResponse(v), e
+}
+func (h *Handler) SearchPeople(ctx context.Context, r gen.SearchPeopleRequestObject) (gen.SearchPeopleResponseObject, error) {
+	id, e := userID(ctx)
+	if e != nil {
+		return nil, e
+	}
+	q := strings.TrimSpace(r.Params.Q)
+	if utf8.RuneCountInString(q) < 2 || utf8.RuneCountInString(q) > 80 {
+		return nil, invalid()
+	}
+	v, e := h.store.people(ctx, id, q)
+	return gen.SearchPeople200JSONResponse(v), e
+}
+func (h *Handler) SocialAction(ctx context.Context, r gen.SocialActionRequestObject) (gen.SocialActionResponseObject, error) {
+	id, e := userID(ctx)
+	if e != nil {
+		return nil, e
+	}
+	if r.Body == nil || r.Body.TargetId == uuid.Nil || r.Body.TargetId == id {
+		return nil, invalid()
+	}
+	if e = h.store.action(ctx, id, *r.Body); e != nil {
+		return nil, e
+	}
+	return gen.SocialAction204Response{}, nil
+}
+func (h *Handler) GetDirectMessages(ctx context.Context, r gen.GetDirectMessagesRequestObject) (gen.GetDirectMessagesResponseObject, error) {
+	id, e := userID(ctx)
+	if e != nil {
+		return nil, e
+	}
+	if r.PeerId == uuid.Nil || r.PeerId == id || (r.Params.Before != nil && *r.Params.Before < 1) {
+		return nil, invalid()
+	}
+	v, e := h.store.messages(ctx, id, r.PeerId, r.Params.Before)
+	return gen.GetDirectMessages200JSONResponse(v), e
+}
+func validateMessage(m gen.SendDirectMessage) error {
+	if m.Id == uuid.Nil || utf8.RuneCountInString(m.Text) > 4000 {
+		return invalid()
+	}
+	switch m.AttachmentKind {
+	case "none":
+		if strings.TrimSpace(m.Text) == "" || m.AttachmentId != nil {
+			return invalid()
+		}
+	case "track", "playlist":
+		if m.AttachmentId == nil || *m.AttachmentId == uuid.Nil {
+			return invalid()
+		}
+	default:
+		return invalid()
+	}
+	return nil
+}
+func (h *Handler) SendDirectMessage(ctx context.Context, r gen.SendDirectMessageRequestObject) (gen.SendDirectMessageResponseObject, error) {
+	id, e := userID(ctx)
+	if e != nil {
+		return nil, e
+	}
+	if r.Body == nil || r.PeerId == id || r.PeerId == uuid.Nil {
+		return nil, invalid()
+	}
+	if e = validateMessage(*r.Body); e != nil {
+		return nil, e
+	}
+	v, e := h.store.send(ctx, id, r.PeerId, *r.Body)
+	return gen.SendDirectMessage200JSONResponse(v), e
+}

services/api/internal/social/social_integration_test.go

@@ -0,0 +1,143 @@
+//go:build integration
+
+package social
+
+import (
+	"context"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	"os"
+	"pipu/api/internal/api/gen"
+	"pipu/api/internal/platform/db"
+	"sync"
+	"testing"
+)
+
+func TestSocialIsolationAndLifecycle(t *testing.T) {
+	url := os.Getenv("TEST_DATABASE_URL")
+	if url == "" {
+		t.Skip("TEST_DATABASE_URL is not set")
+	}
+	ctx := context.Background()
+	pool, e := db.Connect(ctx, url)
+	require.NoError(t, e)
+	defer pool.Close()
+	require.NoError(t, db.Migrate(ctx, pool))
+	s := &Store{pool: pool}
+	a, b, c := uuid.New(), uuid.New(), uuid.New()
+	for _, id := range []uuid.UUID{a, b, c} {
+		_, e = pool.Exec(ctx, `INSERT INTO users(id,username,password_hash) VALUES($1,$2,'x')`, id, id.String())
+		require.NoError(t, e)
+	}
+	defer pool.Exec(ctx, `DELETE FROM users WHERE id=ANY($1)`, []uuid.UUID{a, b, c})
+	action := func(user, peer uuid.UUID, op string) error {
+		return s.action(ctx, user, gen.SocialAction{Action: gen.SocialActionAction(op), TargetId: peer})
+	}
+	require.NoError(t, action(a, b, "request"))
+	require.NoError(t, action(a, b, "request"))
+	parallel(t, func() error { return action(a, b, "request") })
+	inbox, e := s.inbox(ctx, b)
+	require.NoError(t, e)
+	require.Len(t, inbox.Notifications, 1)
+	require.Equal(t, 1, inbox.Unread)
+	require.EqualValues(t, "incoming", inbox.People[0].Relationship)
+	require.Error(t, action(a, b, "accept"))
+	require.Error(t, action(c, b, "accept"))
+	_, e = s.send(ctx, a, b, gen.SendDirectMessage{Id: uuid.New(), Text: "before acceptance", AttachmentKind: "none"})
+	require.Error(t, e)
+	require.NoError(t, action(b, a, "accept"))
+	require.NoError(t, action(b, a, "accept"))
+	inbox, e = s.inbox(ctx, a)
+	require.NoError(t, e)
+	require.Len(t, inbox.Notifications, 1)
+	require.EqualValues(t, "friend", inbox.People[0].Relationship)
+	body := gen.SendDirectMessage{Id: uuid.New(), Text: "Привет", AttachmentKind: "none"}
+	m, e := s.send(ctx, a, b, body)
+	require.NoError(t, e)
+	again, e := s.send(ctx, a, b, body)
+	require.NoError(t, e)
+	require.Equal(t, m.Seq, again.Seq)
+	parallel(t, func() error { _, err := s.send(ctx, a, b, body); return err })
+	body.Text = "changed"
+	_, e = s.send(ctx, a, b, body)
+	require.Error(t, e)
+	_, e = s.messages(ctx, c, b, nil)
+	require.Error(t, e)
+	inbox, e = s.inbox(ctx, b)
+	require.NoError(t, e)
+	require.Equal(t, 1, inbox.Unread)
+	// Another user cannot acknowledge this user's notification.
+	require.NoError(t, s.action(ctx, c, gen.SocialAction{Action: "read_notification", TargetId: inbox.Notifications[0].Id}))
+	inbox, e = s.inbox(ctx, b)
+	require.NoError(t, e)
+	require.Equal(t, 1, inbox.Unread)
+	second, e := s.send(ctx, a, b, gen.SendDirectMessage{Id: uuid.New(), Text: "Later", AttachmentKind: "none"})
+	require.NoError(t, e)
+	require.NoError(t, s.action(ctx, b, gen.SocialAction{Action: "read_chat", TargetId: a, UpToSeq: &m.Seq}))
+	history, e := s.messages(ctx, a, b, nil)
+	require.NoError(t, e)
+	require.Len(t, history, 2)
+	require.True(t, history[0].Read)
+	require.False(t, history[1].Read)
+	before, e := s.messages(ctx, b, a, &second.Seq)
+	require.NoError(t, e)
+	require.Len(t, before, 1)
+	require.NoError(t, action(b, a, "block"))
+	require.NoError(t, action(a, b, "block"))
+	for _, id := range []uuid.UUID{a, b} {
+		blocked, err := s.inbox(ctx, id)
+		require.NoError(t, err)
+		require.Len(t, blocked.People, 1)
+		require.EqualValues(t, "blocked", blocked.People[0].Relationship)
+	}
+	require.NoError(t, action(a, b, "unblock"))
+	_, e = s.send(ctx, a, b, gen.SendDirectMessage{Id: uuid.New(), Text: "blocked", AttachmentKind: "none"})
+	require.Error(t, e)
+	_, e = s.messages(ctx, b, a, nil)
+	require.Error(t, e)
+	require.Error(t, action(a, b, "request"))
+	found, e := s.people(ctx, a, b.String())
+	require.NoError(t, e)
+	require.Empty(t, found)
+	require.NoError(t, action(b, a, "unblock"))
+	require.Error(t, action(a, b, "accept"))
+	require.NoError(t, action(a, b, "request"))
+	require.NoError(t, action(b, a, "accept"))
+	require.NoError(t, action(a, b, "remove"))
+	_, e = s.messages(ctx, b, a, nil)
+	require.Error(t, e)
+	// An attachment must reference available catalog music or an active playlist share.
+	require.NoError(t, action(a, c, "request"))
+	require.NoError(t, action(c, a, "accept"))
+	missing := uuid.New()
+	_, e = s.send(ctx, a, c, gen.SendDirectMessage{Id: uuid.New(), AttachmentKind: "playlist", AttachmentId: &missing})
+	require.Error(t, e)
+	playlist, share := uuid.New(), uuid.New()
+	_, e = pool.Exec(ctx, `INSERT INTO playlists(id,user_id,name,share_id) VALUES($1,$2,'Music',$3)`, playlist, a, share)
+	require.NoError(t, e)
+	attachment, e := s.send(ctx, a, c, gen.SendDirectMessage{Id: uuid.New(), AttachmentKind: "playlist", AttachmentId: &share})
+	require.NoError(t, e)
+	require.Equal(t, &share, attachment.AttachmentId)
+	_, e = pool.Exec(ctx, `UPDATE playlists SET share_id=NULL WHERE id=$1`, playlist)
+	require.NoError(t, e)
+	_, e = s.send(ctx, a, c, gen.SendDirectMessage{Id: uuid.New(), AttachmentKind: "playlist", AttachmentId: &share})
+	require.Error(t, e)
+	found, e = s.people(ctx, a, "%_")
+	require.NoError(t, e)
+	require.Empty(t, found)
+}
+
+func parallel(t *testing.T, operation func() error) {
+	t.Helper()
+	var wg sync.WaitGroup
+	errors := make(chan error, 8)
+	for range 8 {
+		wg.Add(1)
+		go func() { defer wg.Done(); errors <- operation() }()
+	}
+	wg.Wait()
+	close(errors)
+	for err := range errors {
+		require.NoError(t, err)
+	}
+}

services/api/internal/social/social_test.go

@@ -0,0 +1,40 @@
+package social
+
+import (
+	"context"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	"pipu/api/internal/api/gen"
+	"pipu/api/internal/auth"
+	"strings"
+	"testing"
+)
+
+func TestMessageValidation(t *testing.T) {
+	id := uuid.New()
+	for _, m := range []gen.SendDirectMessage{
+		{Id: id, Text: " \n ", AttachmentKind: "none"},
+		{Text: "hello", AttachmentKind: "none"},
+		{Id: id, Text: strings.Repeat("я", 4001), AttachmentKind: "none"},
+		{Id: id, Text: "hello", AttachmentKind: "url", AttachmentId: &id},
+		{Id: id, AttachmentKind: "playlist"},
+		{Id: id, Text: "hello", AttachmentKind: "none", AttachmentId: &id},
+	} {
+		require.Error(t, validateMessage(m))
+	}
+	require.NoError(t, validateMessage(gen.SendDirectMessage{Id: id, Text: strings.Repeat("я", 4000), AttachmentKind: "none"}))
+	require.NoError(t, validateMessage(gen.SendDirectMessage{Id: id, AttachmentKind: "track", AttachmentId: &id}))
+}
+func TestHandlersRequireIdentityAndRejectSelf(t *testing.T) {
+	h := &Handler{}
+	_, e := h.GetSocial(context.Background(), gen.GetSocialRequestObject{})
+	require.Error(t, e)
+	id := uuid.New()
+	ctx := auth.WithPrincipal(context.Background(), auth.Principal{UserID: id})
+	_, e = h.SocialAction(ctx, gen.SocialActionRequestObject{Body: &gen.SocialAction{Action: "request", TargetId: id}})
+	require.Error(t, e)
+	_, e = h.SendDirectMessage(ctx, gen.SendDirectMessageRequestObject{PeerId: id})
+	require.Error(t, e)
+	_, e = h.SearchPeople(ctx, gen.SearchPeopleRequestObject{Params: gen.SearchPeopleParams{Q: " "}})
+	require.Error(t, e)
+}

services/api/internal/social/store.go

@@ -0,0 +1,318 @@
+package social
+
+import (
+	"context"
+	"errors"
+	"strings"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"pipu/api/internal/api/gen"
+	"pipu/api/internal/platform/apperr"
+)
+
+type Store struct{ pool *pgxpool.Pool }
+
+// Every mutation of a pair takes the same transaction lock, including block and send.
+// Thus a send cannot race past a committed unfriend/block, even on another API replica.
+func pairTx(ctx context.Context, pool *pgxpool.Pool, a, b uuid.UUID) (pgx.Tx, error) {
+	tx, e := pool.Begin(ctx)
+	if e != nil {
+		return nil, e
+	}
+	low, high := a.String(), b.String()
+	if low > high {
+		low, high = high, low
+	}
+	_, e = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, low+high)
+	if e != nil {
+		_ = tx.Rollback(ctx)
+		return nil, e
+	}
+	return tx, nil
+}
+func allowed(ctx context.Context, tx pgx.Tx, a, b uuid.UUID) error {
+	var ok bool
+	e := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM friendships f JOIN users u ON u.id=$2
+ WHERE low_id=LEAST($1::uuid,$2::uuid) AND high_id=GREATEST($1::uuid,$2::uuid) AND accepted AND NOT u.is_blocked)
+ AND NOT EXISTS(SELECT 1 FROM social_blocks WHERE (user_id=$1 AND peer_id=$2) OR (user_id=$2 AND peer_id=$1))`, a, b).Scan(&ok)
+	if e != nil {
+		return e
+	}
+	if !ok {
+		return denied()
+	}
+	return nil
+}
+
+const personSQL = `SELECT u.id,u.username,COALESCE(u.display_name,u.username::text),
+ CASE WHEN b.user_id IS NOT NULL THEN 'blocked' WHEN f.accepted THEN 'friend' WHEN f.sender_id=$1 THEN 'outgoing' WHEN f.sender_id IS NOT NULL THEN 'incoming' ELSE 'none' END,
+ (SELECT count(*) FROM direct_messages m WHERE m.recipient_id=$1 AND m.sender_id=u.id AND m.read_at IS NULL)
+ FROM users u
+ LEFT JOIN friendships f ON f.low_id=LEAST($1::uuid,u.id) AND f.high_id=GREATEST($1::uuid,u.id)
+ LEFT JOIN social_blocks b ON b.user_id=$1 AND b.peer_id=u.id
+ WHERE u.id<>$1 AND NOT u.is_blocked
+ AND (b.user_id IS NOT NULL OR NOT EXISTS(SELECT 1 FROM social_blocks x WHERE x.user_id=u.id AND x.peer_id=$1))`
+
+func scanPeople(rows pgx.Rows) ([]gen.SocialPerson, error) {
+	defer rows.Close()
+	out := []gen.SocialPerson{}
+	for rows.Next() {
+		var p gen.SocialPerson
+		if e := rows.Scan(&p.Id, &p.Username, &p.DisplayName, &p.Relationship, &p.Unread); e != nil {
+			return nil, e
+		}
+		out = append(out, p)
+	}
+	return out, rows.Err()
+}
+func (s *Store) people(ctx context.Context, id uuid.UUID, q string) ([]gen.SocialPerson, error) {
+	// strpos treats %, _ and backslashes literally; search is deliberately bounded.
+	rows, e := s.pool.Query(ctx, personSQL+` AND b.user_id IS NULL AND (strpos(lower(u.username::text),lower($2))>0 OR strpos(lower(COALESCE(u.display_name,'')),lower($2))>0) ORDER BY u.username LIMIT 30`, id, q)
+	if e != nil {
+		return nil, e
+	}
+	return scanPeople(rows)
+}
+func (s *Store) inbox(ctx context.Context, id uuid.UUID) (gen.SocialInbox, error) {
+	out := gen.SocialInbox{People: []gen.SocialPerson{}, Notifications: []gen.SocialNotification{}}
+	rows, e := s.pool.Query(ctx, personSQL+` AND (f.sender_id IS NOT NULL OR b.user_id IS NOT NULL) ORDER BY u.username`, id)
+	if e != nil {
+		return out, e
+	}
+	out.People, e = scanPeople(rows)
+	if e != nil {
+		return out, e
+	}
+	rows, e = s.pool.Query(ctx, `SELECT n.id,n.actor_id,COALESCE(u.display_name,u.username::text),n.kind,n.created_at,n.read_at IS NOT NULL
+ FROM social_notifications n JOIN users u ON u.id=n.actor_id WHERE n.user_id=$1 AND NOT u.is_blocked
+ AND NOT EXISTS(SELECT 1 FROM social_blocks b WHERE (b.user_id=$1 AND b.peer_id=n.actor_id) OR (b.peer_id=$1 AND b.user_id=n.actor_id))
+ ORDER BY n.read_at IS NOT NULL,n.created_at DESC,n.id DESC LIMIT 100`, id)
+	if e != nil {
+		return out, e
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var n gen.SocialNotification
+		if e = rows.Scan(&n.Id, &n.ActorId, &n.ActorName, &n.Kind, &n.CreatedAt, &n.Read); e != nil {
+			return out, e
+		}
+		out.Notifications = append(out.Notifications, n)
+	}
+	if e = rows.Err(); e != nil {
+		return out, e
+	}
+	rows.Close()
+	e = s.pool.QueryRow(ctx, `SELECT count(*) FROM social_notifications n JOIN users u ON u.id=n.actor_id WHERE n.user_id=$1 AND n.read_at IS NULL AND NOT u.is_blocked
+ AND NOT EXISTS(SELECT 1 FROM social_blocks b WHERE (b.user_id=$1 AND b.peer_id=n.actor_id) OR (b.peer_id=$1 AND b.user_id=n.actor_id))`, id).Scan(&out.Unread)
+	return out, e
+}
+func notify(ctx context.Context, tx pgx.Tx, to, actor uuid.UUID, kind string, seq *int64) error {
+	_, e := tx.Exec(ctx, `INSERT INTO social_notifications(id,user_id,actor_id,kind,message_seq) VALUES($1,$2,$3,$4,$5)`, uuid.New(), to, actor, kind, seq)
+	return e
+}
+func (s *Store) action(ctx context.Context, id uuid.UUID, a gen.SocialAction) error {
+	peer := a.TargetId
+	if a.Action == "read_notification" {
+		_, e := s.pool.Exec(ctx, `UPDATE social_notifications SET read_at=COALESCE(read_at,now()) WHERE id=$1 AND user_id=$2`, peer, id)
+		return e
+	}
+	tx, e := pairTx(ctx, s.pool, id, peer)
+	if e != nil {
+		return e
+	}
+	defer tx.Rollback(ctx)
+	switch a.Action {
+	case "read_chat":
+		if a.UpToSeq == nil || *a.UpToSeq < 1 {
+			return invalid()
+		}
+		if e = allowed(ctx, tx, id, peer); e != nil {
+			return e
+		}
+		_, e = tx.Exec(ctx, `UPDATE direct_messages SET read_at=COALESCE(read_at,now()) WHERE recipient_id=$1 AND sender_id=$2 AND seq<=$3`, id, peer, *a.UpToSeq)
+		if e == nil {
+			_, e = tx.Exec(ctx, `UPDATE social_notifications SET read_at=COALESCE(read_at,now()) WHERE user_id=$1 AND actor_id=$2 AND message_seq<=$3`, id, peer, *a.UpToSeq)
+		}
+	case "block":
+		_, e = tx.Exec(ctx, `INSERT INTO social_blocks(user_id,peer_id) SELECT $1,id FROM users WHERE id=$2 ON CONFLICT DO NOTHING`, id, peer)
+		if e == nil {
+			_, e = tx.Exec(ctx, `DELETE FROM friendships WHERE low_id=LEAST($1::uuid,$2::uuid) AND high_id=GREATEST($1::uuid,$2::uuid)`, id, peer)
+		}
+		if e == nil {
+			_, e = tx.Exec(ctx, `UPDATE social_notifications SET read_at=COALESCE(read_at,now()) WHERE (user_id=$1 AND actor_id=$2) OR (user_id=$2 AND actor_id=$1)`, id, peer)
+		}
+	case "unblock":
+		_, e = tx.Exec(ctx, `DELETE FROM social_blocks WHERE user_id=$1 AND peer_id=$2`, id, peer)
+	case "remove":
+		_, e = tx.Exec(ctx, `DELETE FROM friendships WHERE low_id=LEAST($1::uuid,$2::uuid) AND high_id=GREATEST($1::uuid,$2::uuid)`, id, peer)
+		if e == nil {
+			_, e = tx.Exec(ctx, `UPDATE social_notifications SET read_at=COALESCE(read_at,now()) WHERE ((user_id=$1 AND actor_id=$2) OR (user_id=$2 AND actor_id=$1)) AND kind IN ('friend_request','friend_accepted')`, id, peer)
+		}
+	case "request", "accept":
+		var available bool
+		e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM users WHERE id=$2 AND NOT is_blocked) AND NOT EXISTS(SELECT 1 FROM social_blocks WHERE (user_id=$1 AND peer_id=$2) OR (user_id=$2 AND peer_id=$1))`, id, peer).Scan(&available)
+		if e != nil {
+			return e
+		}
+		if !available {
+			return denied()
+		}
+		if a.Action == "request" {
+			var exists bool
+			if e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM friendships WHERE low_id=LEAST($1::uuid,$2::uuid) AND high_id=GREATEST($1::uuid,$2::uuid))`, id, peer).Scan(&exists); e != nil {
+				return e
+			}
+			if exists {
+				return tx.Commit(ctx)
+			}
+			// Serialize the sender's quota across requests to different people.
+			if _, e = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, "social-requests:"+id.String()); e != nil {
+				return e
+			}
+			var recent int
+			e = tx.QueryRow(ctx, `SELECT count(*) FROM social_notifications WHERE actor_id=$1 AND kind='friend_request' AND created_at>now()-interval '1 minute'`, id).Scan(&recent)
+			if e != nil {
+				return e
+			}
+			if recent >= 10 {
+				return apperr.TooManyRequests("social.rate_limited", 60)
+			}
+			var count int
+			e = tx.QueryRow(ctx, `SELECT count(*) FROM friendships WHERE sender_id=$1 AND NOT accepted`, id).Scan(&count)
+			if e != nil {
+				return e
+			}
+			if count >= 50 {
+				return apperr.TooManyRequests("social.rate_limited", 60)
+			}
+			result, err := tx.Exec(ctx, `INSERT INTO friendships(low_id,high_id,sender_id) VALUES(LEAST($1::uuid,$2::uuid),GREATEST($1::uuid,$2::uuid),$1) ON CONFLICT DO NOTHING`, id, peer)
+			e = err
+			if e == nil && result.RowsAffected() > 0 {
+				e = notify(ctx, tx, peer, id, "friend_request", nil)
+			}
+		} else {
+			result, err := tx.Exec(ctx, `UPDATE friendships SET accepted=true WHERE low_id=LEAST($1::uuid,$2::uuid) AND high_id=GREATEST($1::uuid,$2::uuid) AND sender_id=$2 AND NOT accepted`, id, peer)
+			e = err
+			if e == nil && result.RowsAffected() > 0 {
+				e = notify(ctx, tx, peer, id, "friend_accepted", nil)
+			} else if e == nil {
+				if e = allowed(ctx, tx, id, peer); e != nil {
+					return e
+				}
+			}
+			if e == nil {
+				_, e = tx.Exec(ctx, `UPDATE social_notifications SET read_at=COALESCE(read_at,now()) WHERE user_id=$1 AND actor_id=$2 AND kind='friend_request'`, id, peer)
+			}
+		}
+	default:
+		return invalid()
+	}
+	if e != nil {
+		return e
+	}
+	return tx.Commit(ctx)
+}
+
+const messageColumns = `id,seq,sender_id,text,attachment_kind,attachment_id,created_at,read_at IS NOT NULL`
+
+func scanMessage(row pgx.Row) (gen.DirectMessage, error) {
+	var m gen.DirectMessage
+	e := row.Scan(&m.Id, &m.Seq, &m.SenderId, &m.Text, &m.AttachmentKind, &m.AttachmentId, &m.CreatedAt, &m.Read)
+	return m, e
+}
+func (s *Store) messages(ctx context.Context, id, peer uuid.UUID, before *int64) ([]gen.DirectMessage, error) {
+	tx, e := pairTx(ctx, s.pool, id, peer)
+	if e != nil {
+		return nil, e
+	}
+	defer tx.Rollback(ctx)
+	if e = allowed(ctx, tx, id, peer); e != nil {
+		return nil, e
+	}
+	rows, e := tx.Query(ctx, `SELECT `+messageColumns+` FROM direct_messages WHERE LEAST(sender_id,recipient_id)=LEAST($1::uuid,$2::uuid) AND GREATEST(sender_id,recipient_id)=GREATEST($1::uuid,$2::uuid) AND ($3::bigint IS NULL OR seq<$3) ORDER BY seq DESC LIMIT 50`, id, peer, before)
+	if e != nil {
+		return nil, e
+	}
+	defer rows.Close()
+	out := []gen.DirectMessage{}
+	for rows.Next() {
+		m, e := scanMessage(rows)
+		if e != nil {
+			return nil, e
+		}
+		out = append(out, m)
+	}
+	if e = rows.Err(); e != nil {
+		return nil, e
+	}
+	for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 {
+		out[i], out[j] = out[j], out[i]
+	}
+	return out, nil
+}
+func (s *Store) send(ctx context.Context, id, peer uuid.UUID, in gen.SendDirectMessage) (gen.DirectMessage, error) {
+	var zero gen.DirectMessage
+	tx, e := pairTx(ctx, s.pool, id, peer)
+	if e != nil {
+		return zero, e
+	}
+	defer tx.Rollback(ctx)
+	if e = allowed(ctx, tx, id, peer); e != nil {
+		return zero, e
+	}
+	// The same client UUID may be retried against different peers concurrently.
+	if _, e = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, "social-message:"+in.Id.String()); e != nil {
+		return zero, e
+	}
+	m, e := scanMessage(tx.QueryRow(ctx, `SELECT `+messageColumns+` FROM direct_messages WHERE id=$1`, in.Id))
+	if e == nil {
+		var recipient uuid.UUID
+		e = tx.QueryRow(ctx, `SELECT recipient_id FROM direct_messages WHERE id=$1`, in.Id).Scan(&recipient)
+		sameAttachment := (m.AttachmentId == nil && in.AttachmentId == nil) || (m.AttachmentId != nil && in.AttachmentId != nil && *m.AttachmentId == *in.AttachmentId)
+		if e != nil {
+			return zero, e
+		}
+		if m.SenderId != id || recipient != peer || m.Text != strings.TrimSpace(in.Text) || string(m.AttachmentKind) != string(in.AttachmentKind) || !sameAttachment {
+			return zero, apperr.Conflict("social.message_conflict", "message ID already used")
+		}
+		return m, nil
+	}
+	if !errors.Is(e, pgx.ErrNoRows) {
+		return zero, e
+	}
+	var count int
+	e = tx.QueryRow(ctx, `SELECT count(*) FROM direct_messages WHERE sender_id=$1 AND recipient_id=$2 AND created_at>now()-interval '1 minute'`, id, peer).Scan(&count)
+	if e != nil {
+		return zero, e
+	}
+	if count >= 30 {
+		return zero, apperr.TooManyRequests("social.rate_limited", 60)
+	}
+	if in.AttachmentKind != "none" {
+		var exists bool
+		query := `SELECT EXISTS(SELECT 1 FROM tracks t JOIN media_files f ON f.id=t.primary_file_id JOIN library_sources s ON s.id=f.source_id WHERE t.id=$1 AND f.missing_since IS NULL AND s.status='mounted')`
+		if in.AttachmentKind == "playlist" {
+			query = `SELECT EXISTS(SELECT 1 FROM playlists WHERE share_id=$1)`
+		}
+		if e = tx.QueryRow(ctx, query, in.AttachmentId).Scan(&exists); e != nil {
+			return zero, e
+		}
+		if !exists {
+			return zero, apperr.NotFound("social.attachment_unavailable", "music is unavailable")
+		}
+	}
+	m, e = scanMessage(tx.QueryRow(ctx, `INSERT INTO direct_messages(id,sender_id,recipient_id,text,attachment_kind,attachment_id) VALUES($1,$2,$3,$4,$5,$6) RETURNING `+messageColumns, in.Id, id, peer, strings.TrimSpace(in.Text), in.AttachmentKind, in.AttachmentId))
+	if e != nil {
+		return zero, e
+	}
+	kind := "message"
+	if in.AttachmentKind != "none" {
+		kind = "music"
+	}
+	if e = notify(ctx, tx, peer, id, kind, &m.Seq); e != nil {
+		return zero, e
+	}
+	return m, tx.Commit(ctx)
+}