feat: disciple link

internal/repository/admin_repo.go

@@ -41,6 +41,7 @@ func (r *AdminRepo) ListUsers(ctx context.Context, q string, limit, offset int)
 		       (SELECT COALESCE(array_agg(oa.provider ORDER BY oa.provider), ARRAY[]::text[])
 		          FROM oauth_accounts oa WHERE oa.user_id = u.id) AS providers,
 		       u.bookmark_limit, u.public_collection_limit, u.private_collection_limit,
+		       u.disciple_id,
 		       COUNT(*) OVER() AS total
 		FROM users u
 		WHERE ($1 = '' OR u.name ILIKE '%'||$1||'%' OR u.handle ILIKE '%'||$1||'%'
@@ -60,7 +61,8 @@ func (r *AdminRepo) ListUsers(ctx context.Context, q string, limit, offset int)
 		if err := rows.Scan(&u.ID, &u.Email, &u.Handle, &u.Name, &u.AvatarURL,
 			&u.Role, &u.Plan, &u.EmailVerifiedAt, &u.BannedAt, &u.PostCount, &u.CreatedAt,
 			&u.HasPassword, &u.Providers,
-			&bookmarkLimit, &publicCollLimit, &privateCollLimit, &total); err != nil {
+			&bookmarkLimit, &publicCollLimit, &privateCollLimit,
+			&u.DiscipleID, &total); err != nil {
 			return nil, 0, fmt.Errorf("scan admin user: %w", err)
 		}
 		// Resolve plan default + overrides so the panel shows what's in force.
@@ -73,6 +75,75 @@ func (r *AdminRepo) ListUsers(ctx context.Context, q string, limit, offset int)
 	return out, total, rows.Err()
 }
 
+// ErrDiscipleClaimed is returned by LinkDisciple when another account already
+// holds the requested disciple — the partial unique index on disciple_id makes
+// the one-to-one rule a database guarantee rather than a check-then-write race.
+var ErrDiscipleClaimed = errors.New("disciple already linked to another user")
+
+// LinkDisciple points a user at a disciple record in Mongo (ObjectID hex).
+// Re-linking simply overwrites any previous value, so "change the link" needs
+// no separate unlink step. Returns ErrDiscipleClaimed when a different account
+// already claims that disciple.
+//
+// `previous` is the link the row held *before* this call (nil when it had
+// none), which is what lets the caller send the welcome email only on the
+// first link — correcting a mis-link shouldn't welcome the devotee twice. The
+// FROM subquery snapshots the old value in the same statement, so there's no
+// read-then-write race.
+func (r *AdminRepo) LinkDisciple(ctx context.Context, userID, discipleID string) (found bool, previous *string, err error) {
+	err = r.pg.Pool.QueryRow(ctx, `
+		UPDATE users u
+		   SET disciple_id = $2,
+		       updated_at = now()
+		  FROM (SELECT id, disciple_id AS old FROM users WHERE id = $1) prev
+		 WHERE u.id = prev.id
+		RETURNING prev.old`, userID, discipleID).Scan(&previous)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return false, nil, nil
+	}
+	if err != nil {
+		if IsUniqueViolation(err) {
+			return false, nil, ErrDiscipleClaimed
+		}
+		return false, nil, fmt.Errorf("admin link disciple: %w", err)
+	}
+	return true, previous, nil
+}
+
+// ClaimDiscipleAnnouncement records that this account's arrival has been
+// announced to the community, returning true only for the caller that actually
+// inserted the row.
+//
+// This is the whole idempotency guard for the broadcast: announcing writes a
+// notification to every user, so it must fire at most once per devotee. Doing
+// it as INSERT … ON CONFLICT DO NOTHING makes the check atomic — a
+// double-clicked button, a retried request, or two admins acting at once still
+// yields exactly one announcement.
+func (r *AdminRepo) ClaimDiscipleAnnouncement(ctx context.Context, userID, discipleID string) (bool, error) {
+	tag, err := r.pg.Pool.Exec(ctx, `
+		INSERT INTO disciple_announcements (user_id, disciple_id)
+		VALUES ($1, $2)
+		ON CONFLICT (user_id) DO NOTHING`, userID, discipleID)
+	if err != nil {
+		return false, fmt.Errorf("claim disciple announcement: %w", err)
+	}
+	return tag.RowsAffected() == 1, nil
+}
+
+// UnlinkDisciple clears a user's disciple link. Idempotent: returns true as
+// long as the user exists, whether or not a link was present.
+func (r *AdminRepo) UnlinkDisciple(ctx context.Context, userID string) (bool, error) {
+	tag, err := r.pg.Pool.Exec(ctx, `
+		UPDATE users
+		   SET disciple_id = NULL,
+		       updated_at = now()
+		 WHERE id = $1`, userID)
+	if err != nil {
+		return false, fmt.Errorf("admin unlink disciple: %w", err)
+	}
+	return tag.RowsAffected() == 1, nil
+}
+
 // SetUserLimits writes the per-user save-limit overrides (nil clears one back
 // to the plan default). Returns false if the id doesn't exist.
 func (r *AdminRepo) SetUserLimits(ctx context.Context, userID string, bookmark, publicCollection, privateCollection *int) (bool, error) {

internal/repository/disciple_repo.go

@@ -2,6 +2,7 @@ package repository
 
 import (
 	"context"
+	"errors"
 	"time"
 
 	"go.mongodb.org/mongo-driver/v2/bson"
@@ -88,6 +89,60 @@ func (r *DiscipleRepo) GetByLetterAndSuffix(ctx context.Context, letter *string,
 	return results, nil
 }
 
+// FindByIDs returns summary docs for the given ObjectID hex strings (order
+// unspecified). Ids that aren't valid ObjectIDs are skipped rather than
+// failing the batch — the ids come from a Postgres column with no FK, so a
+// malformed or stale value must degrade gracefully. An empty/all-invalid
+// input short-circuits without hitting Mongo.
+//
+// Used to hydrate disciple names onto admin user rows: the user→disciple link
+// lives in Postgres while the disciple record lives here.
+func (r *DiscipleRepo) FindByIDs(ctx context.Context, ids []string) ([]DiscipleSummaryDoc, error) {
+	oids := make([]bson.ObjectID, 0, len(ids))
+	for _, id := range ids {
+		if oid, err := bson.ObjectIDFromHex(id); err == nil {
+			oids = append(oids, oid)
+		}
+	}
+	if len(oids) == 0 {
+		return []DiscipleSummaryDoc{}, nil
+	}
+	cursor, err := r.col.Find(ctx,
+		bson.M{"_id": bson.M{"$in": oids}},
+		options.Find().SetProjection(discipleSummaryProjection),
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer cursor.Close(ctx)
+	var results []DiscipleSummaryDoc
+	if err := cursor.All(ctx, &results); err != nil {
+		return nil, err
+	}
+	return results, nil
+}
+
+// FindByID returns one disciple summary, or nil when the id is malformed or
+// no such document exists. Callers use it to validate a link target before
+// writing it to Postgres.
+func (r *DiscipleRepo) FindByID(ctx context.Context, id string) (*DiscipleSummaryDoc, error) {
+	oid, err := bson.ObjectIDFromHex(id)
+	if err != nil {
+		return nil, nil // malformed id — "not found", not an error
+	}
+	var doc DiscipleSummaryDoc
+	err = r.col.FindOne(ctx, bson.M{"_id": oid},
+		options.FindOne().SetProjection(discipleSummaryProjection),
+	).Decode(&doc)
+	if errors.Is(err, mongo.ErrNoDocuments) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &doc, nil
+}
+
 func (r *DiscipleRepo) Count(ctx context.Context) (int64, error) {
 	return r.col.CountDocuments(ctx, bson.M{})
 }

internal/repository/feedback_repo.go

@@ -121,12 +121,19 @@ func (r *FeedbackRepo) ListByUser(ctx context.Context, userID string, limit int)
 // Recent returns the most-recent N feedback rows joined with users
 // for display on /stats. NULL author columns mean anonymous (or the
 // original submitter's account was since deleted).
+//
+// PRIVACY: /stats is a public page, so this query deliberately does NOT select
+// contact_email — only a boolean saying whether one exists. Submitters' email
+// addresses must never reach that template, and not fetching them makes that
+// impossible rather than merely unlikely. Callers that legitimately need the
+// address (the reply/thank-you emails) use FindByID instead.
 func (r *FeedbackRepo) Recent(ctx context.Context, limit int) ([]model.Feedback, error) {
 	if limit <= 0 || limit > 200 {
 		limit = 25
 	}
 	rows, err := r.pg.Pool.Query(ctx, `
-		SELECT f.id, f.user_id, f.category, f.message, f.contact_email, f.created_at,
+		SELECT f.id, f.user_id, f.category, f.message, f.created_at,
+		       (f.contact_email IS NOT NULL AND f.contact_email <> '') AS has_contact_email,
 		       f.acknowledged_at, f.resolved_at,
 		       u.handle, u.name
 		FROM feedback f
@@ -141,7 +148,8 @@ func (r *FeedbackRepo) Recent(ctx context.Context, limit int) ([]model.Feedback,
 	out := []model.Feedback{}
 	for rows.Next() {
 		var f model.Feedback
-		if err := rows.Scan(&f.ID, &f.UserID, &f.Category, &f.Message, &f.ContactEmail, &f.CreatedAt,
+		if err := rows.Scan(&f.ID, &f.UserID, &f.Category, &f.Message, &f.CreatedAt,
+			&f.HasContactEmail,
 			&f.AcknowledgedAt, &f.ResolvedAt,
 			&f.AuthorHandle, &f.AuthorName); err != nil {
 			return nil, fmt.Errorf("scan feedback: %w", err)

internal/repository/social_graph_repo.go

@@ -334,6 +334,62 @@ func (r *NotificationRepo) CreateAdminMessage(ctx context.Context, userID, body
 	return nil
 }
 
+// BroadcastToAllUsers writes one notification of `kind` to every active user
+// except `actorID`, in a single INSERT…SELECT. One round trip regardless of
+// audience size — the alternative (a row per user from Go) would be hundreds
+// of round trips on the request path.
+//
+// Deliberately does NOT filter on notification_prefs: ListByUser already
+// excludes kinds the reader has disabled for the in_app channel at read time,
+// so writing the rows unconditionally keeps a re-enabled preference able to
+// surface history — exactly how the existing kinds behave.
+//
+// Banned accounts are skipped; they can't sign in to read it anyway.
+func (r *NotificationRepo) BroadcastToAllUsers(ctx context.Context, kind, actorID string) (int64, error) {
+	tag, err := r.pg.Pool.Exec(ctx, `
+		INSERT INTO notifications (user_id, actor_id, kind)
+		SELECT u.id, $2, $1
+		FROM users u
+		WHERE u.id <> $2 AND u.banned_at IS NULL`, kind, actorID)
+	if err != nil {
+		return 0, fmt.Errorf("broadcast notification: %w", err)
+	}
+	return tag.RowsAffected(), nil
+}
+
+// PushRecipientsForKind lists the ids of users who should receive a *push* for
+// `kind` — everyone except the actor, minus banned accounts, minus anyone who
+// turned this kind off for the push channel, and minus anyone with no push
+// subscription registered at all.
+//
+// That last filter is what keeps a broadcast cheap: only devotees who actually
+// enabled web push are worth iterating, which is typically a small fraction of
+// the user base.
+func (r *NotificationRepo) PushRecipientsForKind(ctx context.Context, kind, actorID string) ([]string, error) {
+	rows, err := r.pg.Pool.Query(ctx, `
+		SELECT DISTINCT u.id
+		FROM users u
+		JOIN push_subscriptions ps ON ps.user_id = u.id
+		WHERE u.id <> $2 AND u.banned_at IS NULL
+		  AND NOT EXISTS (
+		      SELECT 1 FROM notification_prefs np
+		      WHERE np.user_id = u.id AND np.kind = $1
+		        AND np.channel = 'push' AND np.enabled = FALSE)`, kind, actorID)
+	if err != nil {
+		return nil, fmt.Errorf("list push recipients: %w", err)
+	}
+	defer rows.Close()
+	out := []string{}
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			return nil, fmt.Errorf("scan push recipient: %w", err)
+		}
+		out = append(out, id)
+	}
+	return out, rows.Err()
+}
+
 // 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

internal/repository/user_repo.go

@@ -37,7 +37,8 @@ const userColumns = `id, email, handle, password_hash, name, avatar_url, role, p
 	email_verified_at, token_version, created_at, updated_at, country,
 	sampradaya, gender, initiated, initiated_at, daily_chanting_rounds,
 	preferred_chanting_time, peak_active_time, banned_at,
-	bookmark_limit, public_collection_limit, private_collection_limit`
+	bookmark_limit, public_collection_limit, private_collection_limit,
+	disciple_id`
 
 // rowScanner is satisfied by both pgx.Row (QueryRow) and pgx.Rows, so
 // scanUser works for single-row and iterated reads alike.
@@ -53,6 +54,7 @@ func scanUser(row rowScanner, u *model.User) error {
 		&u.Sampradaya, &u.Gender, &u.Initiated, &u.InitiatedAt, &u.DailyChantingRounds,
 		&u.PreferredChantingTime, &u.PeakActiveTime, &u.BannedAt,
 		&u.BookmarkLimit, &u.PublicCollectionLimit, &u.PrivateCollectionLimit,
+		&u.DiscipleID,
 	)
 }
 
@@ -402,6 +404,41 @@ func (r *UserRepo) AllHandles(ctx context.Context) ([]string, error) {
 	return handles, rows.Err()
 }
 
+// DiscipleLink pairs a Mongo disciple id with the platform account an admin
+// has linked to it. Powers the /disciples browser highlight: the reader page
+// fetches the whole (small) set once and marks the matching cards.
+type DiscipleLink struct {
+	DiscipleID string  `json:"disciple_id"`
+	Handle     string  `json:"handle"`
+	Name       string  `json:"name"`
+	AvatarURL  *string `json:"avatar_url"`
+}
+
+// ListDiscipleLinks returns every user→disciple link. Deliberately unpaged:
+// only Śrīla Prabhupāda's own disciples are ever linked, so this is dozens of
+// rows at most and the partial index on disciple_id keeps it a cheap scan.
+// Suspended accounts are excluded so a banned user can't keep a public badge.
+func (r *UserRepo) ListDiscipleLinks(ctx context.Context) ([]DiscipleLink, error) {
+	rows, err := r.pg.Pool.Query(ctx, `
+		SELECT disciple_id, handle, name, avatar_url
+		FROM users
+		WHERE disciple_id IS NOT NULL AND banned_at IS NULL
+		ORDER BY name`)
+	if err != nil {
+		return nil, fmt.Errorf("list disciple links: %w", err)
+	}
+	defer rows.Close()
+	out := []DiscipleLink{}
+	for rows.Next() {
+		var l DiscipleLink
+		if err := rows.Scan(&l.DiscipleID, &l.Handle, &l.Name, &l.AvatarURL); err != nil {
+			return nil, fmt.Errorf("scan disciple link: %w", err)
+		}
+		out = append(out, l)
+	}
+	return out, rows.Err()
+}
+
 // RecentlyJoined returns up to `limit` users who registered on/after `since`,
 // newest first, as a discovery list for the feed's "new on the platform"
 // strip. Excludes the viewer, anyone the viewer already has a follow edge with