mrleo1nid
cd84ceecb2
ci / api / api (push) Successful in 1m11s
ci / db / db (push) Failing after 23s
ci / mobile / mobile (Metro-бандл Android) (push) Successful in 1m29s
ci / ts / ts (packages + web) (push) Successful in 3m15s
ci / contract / contract (сгенерированный код совпадает с openapi.yaml) (push) Successful in 1m27s
355 lines
14 KiB
Go
355 lines
14 KiB
Go
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),
|
|
COALESCE(f.accepted AND b.user_id IS NULL AND u.social_seen_at>now()-interval '45 seconds',false)
|
|
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, &p.Online); 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
|
|
}
|
|
people, e := scanPeople(rows)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
return people, s.attachActivity(ctx, id, people)
|
|
}
|
|
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
|
|
}
|
|
if e = s.attachActivity(ctx, id, out.People); e != nil {
|
|
return out, e
|
|
}
|
|
if e = s.attachLastMessages(ctx, id, out.People); e != nil {
|
|
return out, e
|
|
}
|
|
if e = s.pool.QueryRow(ctx, `SELECT activity_visibility,friends_recommendations FROM users WHERE id=$1`, id).Scan(&out.ActivityVisibility, &out.FriendsRecommendations); 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,CASE WHEN deleted_at IS NULL THEN text ELSE '' END,attachment_kind,attachment_id,created_at,read_at IS NOT NULL,reply_to,edited_at,deleted_at IS NOT NULL`
|
|
|
|
func scanMessage(row pgx.Row) (gen.DirectMessage, error) {
|
|
var m gen.DirectMessage
|
|
var deleted bool
|
|
e := row.Scan(&m.Id, &m.Seq, &m.SenderId, &m.Text, &m.AttachmentKind, &m.AttachmentId, &m.CreatedAt, &m.Read, &m.ReplyTo, &m.EditedAt, &deleted)
|
|
m.Deleted = &deleted
|
|
return m, e
|
|
}
|
|
func (s *Store) messages(ctx context.Context, id, peer uuid.UUID, before *int64) ([]gen.DirectMessage, error) {
|
|
return s.messagePage(ctx, id, peer, before, "")
|
|
}
|
|
func (s *Store) messagePage(ctx context.Context, id, peer uuid.UUID, before *int64, q string) ([]gen.DirectMessage, error) {
|
|
return s.filteredMessages(ctx, id, peer, before, q, false)
|
|
}
|
|
func (s *Store) filteredMessages(ctx context.Context, id, peer uuid.UUID, before *int64, q string, music bool) ([]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) AND ($4='' OR (deleted_at IS NULL AND strpos(lower(text),lower($4))>0)) AND (NOT $5::boolean OR (deleted_at IS NULL AND attachment_kind IN ('track','playlist'))) ORDER BY seq DESC LIMIT 50`, id, peer, before, q, music)
|
|
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]
|
|
}
|
|
rows.Close()
|
|
return out, enrichMessages(ctx, tx, id, out)
|
|
}
|
|
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
|
|
}
|
|
sameReply := (m.ReplyTo == nil && in.ReplyTo == nil) || (m.ReplyTo != nil && in.ReplyTo != nil && *m.ReplyTo == *in.ReplyTo)
|
|
if m.SenderId != id || recipient != peer || m.Text != strings.TrimSpace(in.Text) || string(m.AttachmentKind) != string(in.AttachmentKind) || !sameAttachment || !sameReply {
|
|
return zero, apperr.Conflict("social.message_conflict", "message ID already used")
|
|
}
|
|
out := []gen.DirectMessage{m}
|
|
e = enrichMessages(ctx, tx, id, out)
|
|
return out[0], e
|
|
}
|
|
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")
|
|
}
|
|
}
|
|
if in.ReplyTo != nil {
|
|
var valid bool
|
|
e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM direct_messages WHERE id=$1 AND LEAST(sender_id,recipient_id)=LEAST($2::uuid,$3::uuid) AND GREATEST(sender_id,recipient_id)=GREATEST($2::uuid,$3::uuid) AND deleted_at IS NULL)`, in.ReplyTo, id, peer).Scan(&valid)
|
|
if e != nil {
|
|
return zero, e
|
|
}
|
|
if !valid {
|
|
return zero, apperr.Conflict("social.message_conflict", "reply unavailable")
|
|
}
|
|
}
|
|
m, e = scanMessage(tx.QueryRow(ctx, `INSERT INTO direct_messages(id,sender_id,recipient_id,text,attachment_kind,attachment_id,reply_to) VALUES($1,$2,$3,$4,$5,$6,$7) RETURNING `+messageColumns, in.Id, id, peer, strings.TrimSpace(in.Text), in.AttachmentKind, in.AttachmentId, in.ReplyTo))
|
|
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)
|
|
}
|