479 lines
18 KiB
Go
479 lines
18 KiB
Go
package repository
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/prabhupadasays/spsays-bff/internal/model"
|
|
)
|
|
|
|
// social_graph_repo.go bundles the four small repos backing the social
|
|
// graph — follows, blocks, reports, and notifications. Each is a thin
|
|
// wrapper over the shared Postgres pool, mirroring the per-domain repo
|
|
// convention used elsewhere (UserRepo, PostRepo, …). They're grouped in
|
|
// one file because each is only a handful of queries and they share the
|
|
// same dependency.
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// FollowRepo
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// FollowRepo manages the request-based follows edge table.
|
|
type FollowRepo struct{ pg *Postgres }
|
|
|
|
func NewFollowRepo(pg *Postgres) *FollowRepo { return &FollowRepo{pg: pg} }
|
|
|
|
// Request inserts a pending follow edge follower→following. Idempotent:
|
|
// a repeated request (the edge already exists, pending or accepted) is a
|
|
// no-op. Returns created=true only when a brand-new pending row was
|
|
// inserted, so the caller fires the "X requested to follow you"
|
|
// notification exactly once.
|
|
func (r *FollowRepo) Request(ctx context.Context, followerID, followingID string) (created bool, err error) {
|
|
tag, err := r.pg.Pool.Exec(ctx, `
|
|
INSERT INTO follows (follower_id, following_id, status)
|
|
VALUES ($1, $2, 'pending')
|
|
ON CONFLICT (follower_id, following_id) DO NOTHING`,
|
|
followerID, followingID,
|
|
)
|
|
if err != nil {
|
|
return false, fmt.Errorf("request follow: %w", err)
|
|
}
|
|
return tag.RowsAffected() == 1, nil
|
|
}
|
|
|
|
// Accept flips a pending edge follower→following to accepted and stamps
|
|
// responded_at. Returns accepted=true only on the transition from
|
|
// pending (so the "X accepted your request" notification fires once); a
|
|
// call on an already-accepted or absent edge returns false.
|
|
func (r *FollowRepo) Accept(ctx context.Context, followerID, followingID string) (accepted bool, err error) {
|
|
tag, err := r.pg.Pool.Exec(ctx, `
|
|
UPDATE follows
|
|
SET status = 'accepted', responded_at = now()
|
|
WHERE follower_id = $1 AND following_id = $2 AND status = 'pending'`,
|
|
followerID, followingID,
|
|
)
|
|
if err != nil {
|
|
return false, fmt.Errorf("accept follow: %w", err)
|
|
}
|
|
return tag.RowsAffected() == 1, nil
|
|
}
|
|
|
|
// Delete removes the edge follower→following regardless of status. Backs
|
|
// unfollow (accepted), cancel-request (own pending outgoing), reject
|
|
// (their pending incoming), and remove-follower (their accepted edge to
|
|
// me). Idempotent.
|
|
func (r *FollowRepo) Delete(ctx context.Context, followerID, followingID string) error {
|
|
_, err := r.pg.Pool.Exec(ctx, `
|
|
DELETE FROM follows WHERE follower_id = $1 AND following_id = $2`,
|
|
followerID, followingID,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("delete follow: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Status returns the status of the edge follower→following ('pending' or
|
|
// 'accepted'), or "" when no edge exists.
|
|
func (r *FollowRepo) Status(ctx context.Context, followerID, followingID string) (string, error) {
|
|
var status string
|
|
err := r.pg.Pool.QueryRow(ctx, `
|
|
SELECT status FROM follows WHERE follower_id = $1 AND following_id = $2`,
|
|
followerID, followingID,
|
|
).Scan(&status)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return "", nil
|
|
}
|
|
return "", fmt.Errorf("follow status: %w", err)
|
|
}
|
|
return status, nil
|
|
}
|
|
|
|
// CountFollowers counts accepted edges pointing at userID.
|
|
// ListFollowers returns the accepted followers of userID (users who follow
|
|
// them), newest first, optionally filtered by a name/handle search `q`.
|
|
// Returns the page plus the total matching count (for pagination).
|
|
func (r *FollowRepo) ListFollowers(ctx context.Context, userID, q string, limit, offset int) ([]model.UserSummary, int, error) {
|
|
return r.listRelations(ctx, "f.following_id = $1", "f.follower_id", userID, q, limit, offset)
|
|
}
|
|
|
|
// ListFollowing returns the users userID follows (accepted), newest first,
|
|
// optionally filtered by `q`. Returns the page plus the total count.
|
|
func (r *FollowRepo) ListFollowing(ctx context.Context, userID, q string, limit, offset int) ([]model.UserSummary, int, error) {
|
|
return r.listRelations(ctx, "f.follower_id = $1", "f.following_id", userID, q, limit, offset)
|
|
}
|
|
|
|
// listRelations powers both directions: `whereOwn` pins the row to userID on
|
|
// one side, `joinCol` is the follows column holding the *other* user (the one
|
|
// we join to users and return). `q` filters by name/handle (ILIKE); empty `q`
|
|
// disables the filter. The column expressions are fixed internal constants —
|
|
// never user input — so this stays injection-safe.
|
|
func (r *FollowRepo) listRelations(
|
|
ctx context.Context, whereOwn, joinCol, userID, q string, limit, offset int,
|
|
) ([]model.UserSummary, int, error) {
|
|
like := "%" + q + "%"
|
|
|
|
var total int
|
|
countSQL := `
|
|
SELECT count(*)
|
|
FROM follows f JOIN users u ON u.id = ` + joinCol + `
|
|
WHERE ` + whereOwn + ` AND f.status = 'accepted'
|
|
AND ($2 = '' OR u.name ILIKE $3 OR u.handle ILIKE $3)`
|
|
if err := r.pg.Pool.QueryRow(ctx, countSQL, userID, q, like).Scan(&total); err != nil {
|
|
return nil, 0, fmt.Errorf("count relations: %w", err)
|
|
}
|
|
if total == 0 {
|
|
return []model.UserSummary{}, 0, nil
|
|
}
|
|
|
|
listSQL := `
|
|
SELECT u.handle, u.name, u.avatar_url
|
|
FROM follows f JOIN users u ON u.id = ` + joinCol + `
|
|
WHERE ` + whereOwn + ` AND f.status = 'accepted'
|
|
AND ($2 = '' OR u.name ILIKE $3 OR u.handle ILIKE $3)
|
|
ORDER BY f.created_at DESC, u.handle ASC
|
|
LIMIT $4 OFFSET $5`
|
|
rows, err := r.pg.Pool.Query(ctx, listSQL, userID, q, like, limit, offset)
|
|
if err != nil {
|
|
return nil, 0, fmt.Errorf("list relations: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
out := make([]model.UserSummary, 0, limit)
|
|
for rows.Next() {
|
|
var s model.UserSummary
|
|
if err := rows.Scan(&s.Handle, &s.Name, &s.AvatarURL); err != nil {
|
|
return nil, 0, fmt.Errorf("scan relation: %w", err)
|
|
}
|
|
out = append(out, s)
|
|
}
|
|
return out, total, rows.Err()
|
|
}
|
|
|
|
func (r *FollowRepo) CountFollowers(ctx context.Context, userID string) (int, error) {
|
|
var n int
|
|
err := r.pg.Pool.QueryRow(ctx, `
|
|
SELECT count(*) FROM follows WHERE following_id = $1 AND status = 'accepted'`,
|
|
userID,
|
|
).Scan(&n)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("count followers: %w", err)
|
|
}
|
|
return n, nil
|
|
}
|
|
|
|
// CountFollowing counts accepted edges originating from userID.
|
|
func (r *FollowRepo) CountFollowing(ctx context.Context, userID string) (int, error) {
|
|
var n int
|
|
err := r.pg.Pool.QueryRow(ctx, `
|
|
SELECT count(*) FROM follows WHERE follower_id = $1 AND status = 'accepted'`,
|
|
userID,
|
|
).Scan(&n)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("count following: %w", err)
|
|
}
|
|
return n, nil
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// BlockRepo
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// BlockRepo manages user_blocks. Block() also tears down any follow
|
|
// edges in either direction, so it runs in a transaction.
|
|
type BlockRepo struct{ pg *Postgres }
|
|
|
|
func NewBlockRepo(pg *Postgres) *BlockRepo { return &BlockRepo{pg: pg} }
|
|
|
|
// Block records blocker→blocked (idempotent) and, in the same tx, deletes
|
|
// any follow edges between the two in either direction — blocking someone
|
|
// should sever the relationship both ways.
|
|
func (r *BlockRepo) Block(ctx context.Context, blockerID, blockedID string) error {
|
|
tx, err := r.pg.Pool.Begin(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("begin block tx: %w", err)
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO user_blocks (blocker_id, blocked_id)
|
|
VALUES ($1, $2)
|
|
ON CONFLICT (blocker_id, blocked_id) DO NOTHING`,
|
|
blockerID, blockedID,
|
|
); err != nil {
|
|
return fmt.Errorf("insert block: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `
|
|
DELETE FROM follows
|
|
WHERE (follower_id = $1 AND following_id = $2)
|
|
OR (follower_id = $2 AND following_id = $1)`,
|
|
blockerID, blockedID,
|
|
); err != nil {
|
|
return fmt.Errorf("sever follows on block: %w", err)
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return fmt.Errorf("commit block: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Unblock removes the blocker→blocked record. Idempotent.
|
|
func (r *BlockRepo) Unblock(ctx context.Context, blockerID, blockedID string) error {
|
|
_, err := r.pg.Pool.Exec(ctx, `
|
|
DELETE FROM user_blocks WHERE blocker_id = $1 AND blocked_id = $2`,
|
|
blockerID, blockedID,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("unblock: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// BlockState reports, in one round-trip, whether a blocked b and whether
|
|
// b blocked a.
|
|
func (r *BlockRepo) BlockState(ctx context.Context, a, b string) (aBlockedB, bBlockedA bool, err error) {
|
|
err = r.pg.Pool.QueryRow(ctx, `
|
|
SELECT
|
|
EXISTS(SELECT 1 FROM user_blocks WHERE blocker_id = $1 AND blocked_id = $2),
|
|
EXISTS(SELECT 1 FROM user_blocks WHERE blocker_id = $2 AND blocked_id = $1)`,
|
|
a, b,
|
|
).Scan(&aBlockedB, &bBlockedA)
|
|
if err != nil {
|
|
return false, false, fmt.Errorf("block state: %w", err)
|
|
}
|
|
return aBlockedB, bBlockedA, nil
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// ReportRepo
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// ReportRepo manages user_reports.
|
|
type ReportRepo struct{ pg *Postgres }
|
|
|
|
func NewReportRepo(pg *Postgres) *ReportRepo { return &ReportRepo{pg: pg} }
|
|
|
|
// Create inserts a user-report row.
|
|
func (r *ReportRepo) Create(ctx context.Context, reporterID, reportedID, category, reason string) (*model.UserReport, error) {
|
|
rep := &model.UserReport{}
|
|
err := r.pg.Pool.QueryRow(ctx, `
|
|
INSERT INTO user_reports (reporter_id, reported_id, category, reason)
|
|
VALUES ($1, $2, $3, $4)
|
|
RETURNING id, reporter_id, reported_id, category, reason, created_at, resolved_at`,
|
|
reporterID, reportedID, category, reason,
|
|
).Scan(&rep.ID, &rep.ReporterID, &rep.ReportedID, &rep.Category, &rep.Reason, &rep.CreatedAt, &rep.ResolvedAt)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create report: %w", err)
|
|
}
|
|
return rep, nil
|
|
}
|
|
|
|
// CreatePostReport inserts a post-report row.
|
|
func (r *ReportRepo) CreatePostReport(ctx context.Context, reporterID, postID, category, reason string) (*model.PostReport, error) {
|
|
rep := &model.PostReport{}
|
|
err := r.pg.Pool.QueryRow(ctx, `
|
|
INSERT INTO post_reports (reporter_id, post_id, category, reason)
|
|
VALUES ($1, $2, $3, $4)
|
|
RETURNING id, reporter_id, post_id, category, reason, created_at, resolved_at`,
|
|
reporterID, postID, category, reason,
|
|
).Scan(&rep.ID, &rep.ReporterID, &rep.PostID, &rep.Category, &rep.Reason, &rep.CreatedAt, &rep.ResolvedAt)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create post report: %w", err)
|
|
}
|
|
return rep, nil
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// NotificationRepo
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// NotificationRepo persists notification rows. Reads (list / unread
|
|
// count / mark-read) will be added when the in-app notification center
|
|
// ships; for now the table is write-only, feeding the email fan-out and
|
|
// standing ready for websocket/push.
|
|
type NotificationRepo struct{ pg *Postgres }
|
|
|
|
func NewNotificationRepo(pg *Postgres) *NotificationRepo { return &NotificationRepo{pg: pg} }
|
|
|
|
// Create inserts a notification for recipient userID, triggered by
|
|
// actorID (nil for system notifications), of the given kind.
|
|
func (r *NotificationRepo) Create(ctx context.Context, userID string, actorID *string, kind string) (*model.Notification, error) {
|
|
n := &model.Notification{}
|
|
err := r.pg.Pool.QueryRow(ctx, `
|
|
INSERT INTO notifications (user_id, actor_id, kind)
|
|
VALUES ($1, $2, $3)
|
|
RETURNING id, user_id, actor_id, kind, created_at, read_at`,
|
|
userID, actorID, kind,
|
|
).Scan(&n.ID, &n.UserID, &n.ActorID, &n.Kind, &n.CreatedAt, &n.ReadAt)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create notification: %w", err)
|
|
}
|
|
return n, nil
|
|
}
|
|
|
|
// CreateAdminMessage persists an admin → user direct message. Unlike the
|
|
// event-driven kinds it has no actor and carries free text in `body`.
|
|
func (r *NotificationRepo) CreateAdminMessage(ctx context.Context, userID, body string) error {
|
|
if _, err := r.pg.Pool.Exec(ctx,
|
|
`INSERT INTO notifications (user_id, kind, body) VALUES ($1, 'admin_message', $2)`,
|
|
userID, body,
|
|
); err != nil {
|
|
return fmt.Errorf("create admin message notification: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ListByUser returns a recipient's notifications, newest first, with the
|
|
// actor's display fields joined in for rendering (nil for system
|
|
// notifications or since-deleted actors). Kinds the user has turned off for
|
|
// the in-app channel are filtered at read time — the rows still persist (they
|
|
// back like-aggregation and reappear if the pref is re-enabled).
|
|
func (r *NotificationRepo) ListByUser(ctx context.Context, userID string, limit, offset int) ([]model.Notification, error) {
|
|
rows, err := r.pg.Pool.Query(ctx, `
|
|
SELECT n.id, n.user_id, n.actor_id, n.kind, n.post_id, n.actor_count,
|
|
n.created_at, n.read_at,
|
|
u.name, u.handle, u.avatar_url,
|
|
pu.handle, n.body
|
|
FROM notifications n
|
|
LEFT JOIN users u ON u.id = n.actor_id
|
|
LEFT JOIN posts p ON p.id = n.post_id
|
|
LEFT JOIN users pu ON pu.id = p.user_id
|
|
WHERE n.user_id = $1
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM notification_prefs np
|
|
WHERE np.user_id = n.user_id AND np.kind = n.kind
|
|
AND np.channel = 'in_app' AND np.enabled = FALSE)
|
|
ORDER BY n.created_at DESC
|
|
LIMIT $2 OFFSET $3`, userID, limit, offset,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list notifications: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
out := []model.Notification{}
|
|
for rows.Next() {
|
|
var n model.Notification
|
|
if err := rows.Scan(&n.ID, &n.UserID, &n.ActorID, &n.Kind, &n.PostID, &n.ActorCount,
|
|
&n.CreatedAt, &n.ReadAt,
|
|
&n.ActorName, &n.ActorHandle, &n.ActorAvatar,
|
|
&n.PostAuthorHandle, &n.Body); err != nil {
|
|
return nil, fmt.Errorf("scan notification: %w", err)
|
|
}
|
|
out = append(out, n)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// UpsertLike records (or aggregates) a like notification: one row per
|
|
// (recipient, post). On a repeat like it refreshes the most-recent actor,
|
|
// bumps actor_count, and re-marks the row unread + most-recent.
|
|
//
|
|
// It also throttles the aggregated like *email*: like_email_sent_at is stamped
|
|
// to now() on the first like and again only once `throttle` has elapsed since
|
|
// the last email. Returns the new actor_count and whether an email is due this
|
|
// call (i.e. the stamp was (re)set now) — decided atomically in one statement
|
|
// to avoid double-emailing under concurrent likes.
|
|
func (r *NotificationRepo) UpsertLike(ctx context.Context, userID, actorID, postID string, throttle time.Duration) (count int, emailDue bool, err error) {
|
|
secs := int(throttle.Seconds())
|
|
err = r.pg.Pool.QueryRow(ctx, `
|
|
INSERT INTO notifications (user_id, actor_id, kind, post_id, actor_count, like_email_sent_at)
|
|
VALUES ($1, $2, 'post_like', $3, 1, now())
|
|
ON CONFLICT (user_id, post_id) WHERE kind = 'post_like'
|
|
DO UPDATE SET actor_id = EXCLUDED.actor_id,
|
|
actor_count = notifications.actor_count + 1,
|
|
read_at = NULL,
|
|
created_at = now(),
|
|
like_email_sent_at = CASE
|
|
WHEN notifications.like_email_sent_at IS NULL
|
|
OR notifications.like_email_sent_at < now() - make_interval(secs => $4)
|
|
THEN now()
|
|
ELSE notifications.like_email_sent_at
|
|
END
|
|
RETURNING actor_count, (like_email_sent_at = now())`,
|
|
userID, actorID, postID, secs,
|
|
).Scan(&count, &emailDue)
|
|
if err != nil {
|
|
return 0, false, fmt.Errorf("upsert like notification: %w", err)
|
|
}
|
|
return count, emailDue, nil
|
|
}
|
|
|
|
// CreateMention persists a "you were mentioned" notification linked to the
|
|
// post. Unlike likes there's no aggregation — a mention is a deliberate,
|
|
// low-volume event.
|
|
func (r *NotificationRepo) CreateMention(ctx context.Context, userID, actorID, postID string) error {
|
|
if _, err := r.pg.Pool.Exec(ctx,
|
|
`INSERT INTO notifications (user_id, actor_id, kind, post_id)
|
|
VALUES ($1, $2, 'post_mention', $3)`,
|
|
userID, actorID, postID,
|
|
); err != nil {
|
|
return fmt.Errorf("create mention notification: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// CreateComment persists a comment/reply notification linked to the post.
|
|
// kind is 'post_comment' (to the post author) or 'comment_reply' (to the
|
|
// replied-to comment's author). Like mentions, no aggregation.
|
|
func (r *NotificationRepo) CreateComment(ctx context.Context, userID, actorID, postID, kind string) error {
|
|
if _, err := r.pg.Pool.Exec(ctx,
|
|
`INSERT INTO notifications (user_id, actor_id, kind, post_id)
|
|
VALUES ($1, $2, $3, $4)`,
|
|
userID, actorID, kind, postID,
|
|
); err != nil {
|
|
return fmt.Errorf("create comment notification: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// PostAuthorHandle returns the handle of the user who owns the post, for
|
|
// building a deep-link. Empty string (no error) when the post is gone.
|
|
func (r *NotificationRepo) PostAuthorHandle(ctx context.Context, postID string) (string, error) {
|
|
var h string
|
|
err := r.pg.Pool.QueryRow(ctx,
|
|
`SELECT u.handle FROM posts p JOIN users u ON u.id = p.user_id WHERE p.id = $1`,
|
|
postID).Scan(&h)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return "", nil
|
|
}
|
|
if err != nil {
|
|
return "", fmt.Errorf("post author handle: %w", err)
|
|
}
|
|
return h, nil
|
|
}
|
|
|
|
// CountUnread returns how many of a recipient's notifications are unread,
|
|
// excluding kinds the user has turned off for the in-app channel (matching
|
|
// ListByUser, so the badge never counts rows the feed won't show).
|
|
func (r *NotificationRepo) CountUnread(ctx context.Context, userID string) (int, error) {
|
|
var n int
|
|
err := r.pg.Pool.QueryRow(ctx, `
|
|
SELECT count(*) FROM notifications n
|
|
WHERE n.user_id = $1 AND n.read_at IS NULL
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM notification_prefs np
|
|
WHERE np.user_id = n.user_id AND np.kind = n.kind
|
|
AND np.channel = 'in_app' AND np.enabled = FALSE)`, userID,
|
|
).Scan(&n)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("count unread notifications: %w", err)
|
|
}
|
|
return n, nil
|
|
}
|
|
|
|
// MarkAllRead stamps read_at on every still-unread notification for the
|
|
// user. Idempotent.
|
|
func (r *NotificationRepo) MarkAllRead(ctx context.Context, userID string) error {
|
|
_, err := r.pg.Pool.Exec(ctx,
|
|
`UPDATE notifications SET read_at = now() WHERE user_id = $1 AND read_at IS NULL`, userID,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("mark notifications read: %w", err)
|
|
}
|
|
return nil
|
|
}
|