fix: no attachment broadcast fix

internal/handler/broadcast.go

@@ -0,0 +1,151 @@
+package handler
+
+import (
+	"encoding/json"
+	"errors"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+	"github.com/prabhupadasays/spsays-bff/internal/middleware"
+	"github.com/prabhupadasays/spsays-bff/internal/repository"
+	"github.com/prabhupadasays/spsays-bff/internal/service"
+	"github.com/prabhupadasays/spsays-bff/pkg/response"
+)
+
+// BroadcastHandler serves both sides of announcements: the admin routes that
+// compose and publish them, and the two app routes that fetch and dismiss.
+type BroadcastHandler struct {
+	svc *service.BroadcastService
+}
+
+func NewBroadcastHandler(svc *service.BroadcastService) *BroadcastHandler {
+	return &BroadcastHandler{svc: svc}
+}
+
+func writeBroadcastError(w http.ResponseWriter, err error) {
+	switch {
+	case errors.Is(err, repository.ErrBroadcastGone):
+		response.Error(w, http.StatusNotFound, "broadcast not found")
+	case errors.Is(err, service.ErrBroadcastTitle),
+		errors.Is(err, service.ErrBroadcastBody),
+		errors.Is(err, service.ErrBroadcastCTA),
+		errors.Is(err, service.ErrBroadcastBanner):
+		response.Error(w, http.StatusBadRequest, err.Error())
+	default:
+		response.Error(w, http.StatusInternalServerError, "broadcast request failed")
+	}
+}
+
+// ── App routes ────────────────────────────────────────────────────────────
+
+// Pending — GET /broadcasts/pending. Returns null when nothing is waiting, so
+// the client can call it on load and after a nudge without special-casing.
+func (h *BroadcastHandler) Pending(w http.ResponseWriter, r *http.Request) {
+	b, err := h.svc.Pending(r.Context(), middleware.UserID(r.Context()))
+	if err != nil {
+		writeBroadcastError(w, err)
+		return
+	}
+	response.JSON(w, http.StatusOK, b)
+}
+
+// Seen — POST /broadcasts/{id}/seen. The "at most once" half; idempotent.
+func (h *BroadcastHandler) Seen(w http.ResponseWriter, r *http.Request) {
+	if err := h.svc.MarkSeen(r.Context(), chi.URLParam(r, "id"),
+		middleware.UserID(r.Context())); err != nil {
+		writeBroadcastError(w, err)
+		return
+	}
+	response.NoContent(w)
+}
+
+// ── Admin routes ──────────────────────────────────────────────────────────
+
+type createBroadcastReq struct {
+	Title     string  `json:"title"`
+	Body      string  `json:"body"`
+	BannerKey *string `json:"banner_key"`
+	CTALabel  *string `json:"cta_label"`
+	CTAURL    *string `json:"cta_url"`
+	// ExpiresAt stops the announcement being offered. Without one, someone
+	// signing up months later would still be shown it.
+	ExpiresAt *time.Time `json:"expires_at"`
+	// Publish immediately rather than saving a draft.
+	Publish bool `json:"publish"`
+}
+
+// Create — POST /admin/broadcasts
+func (h *BroadcastHandler) Create(w http.ResponseWriter, r *http.Request) {
+	var req createBroadcastReq
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		response.Error(w, http.StatusBadRequest, "invalid JSON body")
+		return
+	}
+	b, err := h.svc.Create(r.Context(), middleware.UserID(r.Context()), service.CreateParams{
+		Title: req.Title, Body: req.Body, BannerKey: req.BannerKey,
+		CTALabel: req.CTALabel, CTAURL: req.CTAURL, ExpiresAt: req.ExpiresAt,
+	})
+	if err != nil {
+		writeBroadcastError(w, err)
+		return
+	}
+	if req.Publish {
+		published, err := h.svc.Publish(r.Context(), b.ID)
+		if err != nil {
+			writeBroadcastError(w, err)
+			return
+		}
+		b = published
+	}
+	response.JSON(w, http.StatusCreated, b)
+}
+
+// Publish — POST /admin/broadcasts/{id}/publish
+func (h *BroadcastHandler) Publish(w http.ResponseWriter, r *http.Request) {
+	b, err := h.svc.Publish(r.Context(), chi.URLParam(r, "id"))
+	if err != nil {
+		writeBroadcastError(w, err)
+		return
+	}
+	response.JSON(w, http.StatusOK, b)
+}
+
+// List — GET /admin/broadcasts
+func (h *BroadcastHandler) List(w http.ResponseWriter, r *http.Request) {
+	limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
+	items, err := h.svc.List(r.Context(), limit)
+	if err != nil {
+		writeBroadcastError(w, err)
+		return
+	}
+	response.JSON(w, http.StatusOK, items)
+}
+
+// Delete — DELETE /admin/broadcasts/{id}
+func (h *BroadcastHandler) Delete(w http.ResponseWriter, r *http.Request) {
+	if err := h.svc.Delete(r.Context(), chi.URLParam(r, "id")); err != nil {
+		writeBroadcastError(w, err)
+		return
+	}
+	response.NoContent(w)
+}
+
+// PresignBanner — POST /admin/broadcasts/banner-presign
+func (h *BroadcastHandler) PresignBanner(w http.ResponseWriter, r *http.Request) {
+	var req struct {
+		ContentType string `json:"content_type"`
+	}
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		response.Error(w, http.StatusBadRequest, "invalid JSON body")
+		return
+	}
+	upload, key, err := h.svc.PresignBanner(r.Context(), strings.TrimSpace(req.ContentType))
+	if err != nil {
+		response.Error(w, http.StatusBadRequest, err.Error())
+		return
+	}
+	response.JSON(w, http.StatusOK, map[string]string{"upload_url": upload, "key": key})
+}

internal/handler/chat.go

@@ -36,6 +36,10 @@ func writeChatError(w http.ResponseWriter, err error) {
 		response.Error(w, http.StatusNotFound, "conversation not found")
 	case errors.Is(err, service.ErrAvatarUnset):
 		response.Error(w, http.StatusNotFound, err.Error())
+	case errors.Is(err, service.ErrChatBadEntity):
+		response.Error(w, http.StatusBadRequest, err.Error())
+	case errors.Is(err, service.ErrChatBadQuote):
+		response.Error(w, http.StatusBadRequest, err.Error())
 	case errors.Is(err, service.ErrChatBadAvatarKey):
 		response.Error(w, http.StatusBadRequest, err.Error())
 	case errors.Is(err, repository.ErrMessageNotHere):
@@ -92,6 +96,17 @@ type openDMReq struct {
 	Handle string `json:"handle"`
 }
 
+// Unread — GET /chat/unread. Drives the nav badge, so it stays a single
+// count rather than anything the caller has to aggregate.
+func (h *ChatHandler) Unread(w http.ResponseWriter, r *http.Request) {
+	n, err := h.svc.UnreadCount(r.Context(), middleware.UserID(r.Context()))
+	if err != nil {
+		writeChatError(w, err)
+		return
+	}
+	response.JSON(w, http.StatusOK, map[string]int{"count": n})
+}
+
 // OpenDM — POST /chat/conversations/dm. Idempotent: returns the existing
 // thread when there is one, so the client can call it every time it opens a
 // profile without checking first.
@@ -186,7 +201,13 @@ type sendReq struct {
 	// when they dismissed it. Only the URL — never the card's contents.
 	PreviewURL  *string    `json:"preview_url"`
 	Attachments []mediaReq `json:"attachments"`
-	ReplyToID   *string    `json:"reply_to_id"`
+	// quote is a passage shared from the library. `path` must be app-relative;
+	// the service refuses anything that could navigate off-site.
+	Quote *quoteReq `json:"quote"`
+	// entity shares something else in the app — a post — as a card. Only the
+	// reference; the BFF resolves the contents and re-checks visibility.
+	Entity    *entityReq `json:"entity"`
+	ReplyToID *string    `json:"reply_to_id"`
 	// forward_from_id is deliberately NOT accepted yet: forwarding lands with
 	// its own "can the caller see the source?" check in a later phase, and
 	// accepting an unvalidated message reference in the meantime would let a
@@ -210,11 +231,23 @@ func (h *ChatHandler) Send(w http.ResponseWriter, r *http.Request) {
 			FileName: a.FileName,
 		})
 	}
+	var quote *service.QuoteAttachment
+	if req.Quote != nil {
+		quote = &service.QuoteAttachment{
+			Path: req.Quote.Path, Title: req.Quote.Title, Text: req.Quote.Text,
+		}
+	}
+	var entity *service.EntityAttachment
+	if req.Entity != nil {
+		entity = &service.EntityAttachment{Type: req.Entity.Type, ID: req.Entity.ID}
+	}
 	msg, err := h.svc.Send(r.Context(), userID, convID, service.SendRequest{
 		Body:        req.Body,
 		ClientMsgID: req.ClientMsgID,
 		ReplyToID:   req.ReplyToID,
 		PreviewURL:  req.PreviewURL,
+		Quote:       quote,
+		Entity:      entity,
 		Attachments: media,
 	})
 	if err != nil {
@@ -554,6 +587,17 @@ func (h *ChatHandler) UpdateGroup(w http.ResponseWriter, r *http.Request) {
 	response.NoContent(w)
 }
 
+type entityReq struct {
+	Type string `json:"type"`
+	ID   string `json:"id"`
+}
+
+type quoteReq struct {
+	Path  string `json:"path"`
+	Title string `json:"title"`
+	Text  string `json:"text"`
+}
+
 type receiptReq struct {
 	Seq int64 `json:"seq"`
 }

internal/handler/soketi_webhook.go

@@ -1,6 +1,7 @@
 package handler
 
 import (
+	"context"
 	"crypto/hmac"
 	"crypto/sha256"
 	"encoding/hex"
@@ -41,11 +42,29 @@ type SoketiWebhookHandler struct {
 	appKey    string
 	appSecret string
 	feed      *repository.FeedRepo
-	logger    *zap.Logger
+	// chat records presence so a message doesn't push to someone who is
+	// already looking at the app. Optional.
+	chat   ChatPresence
+	logger *zap.Logger
 }
 
-func NewSoketiWebhookHandler(appKey, appSecret string, feed *repository.FeedRepo, logger *zap.Logger) *SoketiWebhookHandler {
+// ChatPresence is the slice of ChatService this handler needs — narrow so the
-	return &SoketiWebhookHandler{appKey: appKey, appSecret: appSecret, feed: feed, logger: logger}
+// webhook doesn't depend on the whole chat surface.
+type ChatPresence interface {
+	MarkPresence(ctx context.Context, userID string, online bool)
+}
+
+func (h *SoketiWebhookHandler) presence(ctx context.Context, userID string, online bool) {
+	if h.chat == nil {
+		return
+	}
+	h.chat.MarkPresence(ctx, userID, online)
+}
+
+func NewSoketiWebhookHandler(appKey, appSecret string, feed *repository.FeedRepo, chat ChatPresence, logger *zap.Logger) *SoketiWebhookHandler {
+	return &SoketiWebhookHandler{
+		appKey: appKey, appSecret: appSecret, feed: feed, chat: chat, logger: logger,
+	}
 }
 
 type soketiWebhookBody struct {
@@ -91,12 +110,20 @@ func (h *SoketiWebhookHandler) Handle(w http.ResponseWriter, r *http.Request) {
 	ctx := r.Context()
 	cleared := 0
 	for _, e := range payload.Events {
-		// A per-user notifications channel emptying = that user is offline.
+		uid, ok := userIDFromNotifChannel(e.Channel)
-		if e.Name != "channel_vacated" {
+		if !ok {
 			continue
 		}
-		if uid, ok := userIDFromNotifChannel(e.Channel); ok {
+		switch e.Name {
+		case "channel_occupied":
+			// This user has a live connection, so chat push is suppressed
+			// for them — they will see the message in the app.
+			h.presence(ctx, uid, true)
+		case "channel_vacated":
+			// Their last tab closed: they are offline. Evict the feed
+			// seen-set and let chat push resume.
 			h.feed.Reset(ctx, uid)
+			h.presence(ctx, uid, false)
 			cleared++
 		}
 	}

internal/model/broadcast.go

@@ -0,0 +1,40 @@
+package model
+
+import (
+	"encoding/json"
+	"time"
+)
+
+// Broadcast is an announcement shown to every user exactly once.
+//
+// Deliberately not a notification: it creates no tray rows and no per-user
+// records until someone actually dismisses it. The modal is the delivery
+// surface; push and the realtime nudge only tell clients to go look.
+type Broadcast struct {
+	ID    string `json:"id"`
+	Title string `json:"title"`
+	Body  string `json:"body"`
+	// Entities are offset/length formatting spans, the same shape chat uses,
+	// so no HTML is ever stored or rendered.
+	Entities json.RawMessage `json:"entities"`
+	// BannerURL is resolved from the stored R2 key for the client.
+	BannerURL *string `json:"banner_url,omitempty"`
+	CTALabel  *string `json:"cta_label,omitempty"`
+	CTAURL    *string `json:"cta_url,omitempty"`
+
+	PublishedAt *time.Time `json:"published_at,omitempty"`
+	ExpiresAt   *time.Time `json:"expires_at,omitempty"`
+	CreatedAt   time.Time  `json:"created_at"`
+
+	// SeenCount is filled for the admin list only — how many people have
+	// dismissed it so far.
+	SeenCount int64 `json:"seen_count,omitempty"`
+}
+
+// Live reports whether the broadcast is currently deliverable.
+func (b *Broadcast) Live(now time.Time) bool {
+	if b.PublishedAt == nil || b.PublishedAt.After(now) {
+		return false
+	}
+	return b.ExpiresAt == nil || b.ExpiresAt.After(now)
+}

internal/model/chat.go

@@ -167,6 +167,15 @@ const (
 	AttachmentFile   = "file"
 	AttachmentLink   = "link"
 	AttachmentEntity = "entity"
+	// AttachmentQuote is a passage shared from the library: url is an
+	// app-relative deep link back to the phrase, title the source reference,
+	// description the passage itself.
+	AttachmentQuote = "quote"
+)
+
+// Entity types a message can reference.
+const (
+	EntityPost = "post"
 )
 
 // Attachment is a media file, link preview or shared entity on a message.
@@ -197,6 +206,16 @@ type Attachment struct {
 	SiteName    *string `json:"site_name,omitempty"`
 	ImageURL    *string `json:"image_url,omitempty"`
 
+	// Shared entity (a post, for now). EntityType/EntityID are the durable
+	// reference; Entity is resolved per viewer at read time, so the same row
+	// renders differently — or not at all — depending on who is looking.
+	EntityType *string         `json:"entity_type,omitempty"`
+	EntityID   *string         `json:"entity_id,omitempty"`
+	Entity     json.RawMessage `json:"entity,omitempty"`
+	// EntityUnavailable is set when the entity was deleted, hidden, or the
+	// viewer has since lost access. The client renders a tombstone.
+	EntityUnavailable bool `json:"entity_unavailable,omitempty"`
+
 	CreatedAt time.Time `json:"created_at"`
 }
 

internal/repository/broadcast_repo.go

@@ -0,0 +1,151 @@
+package repository
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/prabhupadasays/spsays-bff/internal/model"
+)
+
+var ErrBroadcastGone = errors.New("broadcast not found")
+
+type BroadcastRepo struct{ pg *Postgres }
+
+func NewBroadcastRepo(pg *Postgres) *BroadcastRepo { return &BroadcastRepo{pg: pg} }
+
+const broadcastColumns = `id, title, body, entities, banner_key, cta_label,
+	cta_url, published_at, expires_at, created_at`
+
+// scanBroadcast decodes a row produced by broadcastColumns. bannerKey stays
+// out of the model — the service turns it into a URL.
+func scanBroadcast(row pgx.Row, b *model.Broadcast, bannerKey **string) error {
+	return row.Scan(&b.ID, &b.Title, &b.Body, &b.Entities, bannerKey,
+		&b.CTALabel, &b.CTAURL, &b.PublishedAt, &b.ExpiresAt, &b.CreatedAt)
+}
+
+type CreateBroadcastParams struct {
+	Title     string
+	Body      string
+	Entities  []byte
+	BannerKey *string
+	CTALabel  *string
+	CTAURL    *string
+	ExpiresAt *time.Time
+	CreatedBy string
+}
+
+func (r *BroadcastRepo) Create(ctx context.Context, p CreateBroadcastParams) (*model.Broadcast, *string, error) {
+	var b model.Broadcast
+	var key *string
+	err := scanBroadcast(r.pg.Pool.QueryRow(ctx, `
+		INSERT INTO broadcasts (title, body, entities, banner_key, cta_label,
+		                        cta_url, expires_at, created_by)
+		VALUES ($1,$2,$3,$4,$5,$6,$7,$8)
+		RETURNING `+broadcastColumns,
+		p.Title, p.Body, p.Entities, p.BannerKey, p.CTALabel, p.CTAURL,
+		p.ExpiresAt, p.CreatedBy), &b, &key)
+	if err != nil {
+		return nil, nil, fmt.Errorf("create broadcast: %w", err)
+	}
+	return &b, key, nil
+}
+
+// Publish stamps published_at, making the broadcast deliverable. Idempotent:
+// re-publishing keeps the original timestamp so the live window doesn't shift.
+func (r *BroadcastRepo) Publish(ctx context.Context, id string) (*model.Broadcast, *string, error) {
+	var b model.Broadcast
+	var key *string
+	err := scanBroadcast(r.pg.Pool.QueryRow(ctx, `
+		UPDATE broadcasts
+		SET published_at = COALESCE(published_at, now()), updated_at = now()
+		WHERE id = $1
+		RETURNING `+broadcastColumns, id), &b, &key)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, nil, ErrBroadcastGone
+	}
+	if err != nil {
+		return nil, nil, fmt.Errorf("publish broadcast: %w", err)
+	}
+	return &b, key, nil
+}
+
+func (r *BroadcastRepo) Delete(ctx context.Context, id string) error {
+	tag, err := r.pg.Pool.Exec(ctx, `DELETE FROM broadcasts WHERE id = $1`, id)
+	if err != nil {
+		return fmt.Errorf("delete broadcast: %w", err)
+	}
+	if tag.RowsAffected() == 0 {
+		return ErrBroadcastGone
+	}
+	return nil
+}
+
+// Pending returns the newest live broadcast this user has not dismissed.
+func (r *BroadcastRepo) Pending(ctx context.Context, userID string) (*model.Broadcast, *string, error) {
+	var b model.Broadcast
+	var key *string
+	err := scanBroadcast(r.pg.Pool.QueryRow(ctx, `
+		SELECT `+broadcastColumns+`
+		FROM broadcasts b
+		WHERE b.published_at IS NOT NULL
+		  AND b.published_at <= now()
+		  AND (b.expires_at IS NULL OR b.expires_at > now())
+		  AND NOT EXISTS (
+		      SELECT 1 FROM broadcast_views v
+		      WHERE v.broadcast_id = b.id AND v.user_id = $1)
+		-- Newest first. Announcements are time-sensitive, so a stale one
+		-- must not gate a fresh one; dismissing this reveals the next, so
+		-- nothing is skipped either way — only the order differs.
+		ORDER BY b.published_at DESC
+		LIMIT 1`, userID), &b, &key)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, nil, nil // nothing waiting is not an error
+	}
+	if err != nil {
+		return nil, nil, fmt.Errorf("pending broadcast: %w", err)
+	}
+	return &b, key, nil
+}
+
+// MarkSeen records a dismissal. Idempotent — two tabs can race.
+func (r *BroadcastRepo) MarkSeen(ctx context.Context, id, userID string) error {
+	_, err := r.pg.Pool.Exec(ctx, `
+		INSERT INTO broadcast_views (broadcast_id, user_id) VALUES ($1,$2)
+		ON CONFLICT DO NOTHING`, id, userID)
+	if err != nil {
+		return fmt.Errorf("mark broadcast seen: %w", err)
+	}
+	return nil
+}
+
+// List returns broadcasts newest first, with dismissal counts, for the admin.
+func (r *BroadcastRepo) List(ctx context.Context, limit int) ([]model.Broadcast, []*string, error) {
+	rows, err := r.pg.Pool.Query(ctx, `
+		SELECT `+broadcastColumns+`,
+		       (SELECT count(*) FROM broadcast_views v WHERE v.broadcast_id = b.id)
+		FROM broadcasts b
+		ORDER BY COALESCE(b.published_at, b.created_at) DESC
+		LIMIT $1`, limit)
+	if err != nil {
+		return nil, nil, fmt.Errorf("list broadcasts: %w", err)
+	}
+	defer rows.Close()
+
+	out := []model.Broadcast{}
+	keys := []*string{}
+	for rows.Next() {
+		var b model.Broadcast
+		var key *string
+		if err := rows.Scan(&b.ID, &b.Title, &b.Body, &b.Entities, &key,
+			&b.CTALabel, &b.CTAURL, &b.PublishedAt, &b.ExpiresAt, &b.CreatedAt,
+			&b.SeenCount); err != nil {
+			return nil, nil, fmt.Errorf("scan broadcast: %w", err)
+		}
+		out = append(out, b)
+		keys = append(keys, key)
+	}
+	return out, keys, rows.Err()
+}

internal/repository/chat_repo.go

@@ -547,6 +547,12 @@ type AddAttachmentParams struct {
 	Description *string
 	SiteName    *string
 	ImageURL    *string
+
+	// Shared entity: the durable reference plus a snapshot used only as a
+	// fallback. Live data is re-resolved per viewer at read time.
+	EntityType *string
+	EntityID   *string
+	Snapshot   []byte
 }
 
 // AddAttachment persists one attachment. conversation_id is denormalized here
@@ -557,17 +563,20 @@ func (r *ChatRepo) AddAttachment(ctx context.Context, p AddAttachmentParams) (*m
 		INSERT INTO message_attachments
 		    (message_id, conversation_id, ord, kind, r2_key, mime, size_bytes,
 		     width, height, duration_ms, blurhash, file_name,
-		     url, title, description, site_name, image_url)
+		     url, title, description, site_name, image_url,
-		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17)
+		     entity_type, entity_id, snapshot)
+		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20)
 		RETURNING id, message_id, ord, kind, mime, size_bytes, width, height,
 		          duration_ms, blurhash, file_name, url, title, description,
-		          site_name, image_url, created_at`,
+		          site_name, image_url, entity_type, entity_id, snapshot, created_at`,
 		p.MessageID, p.ConversationID, p.Ord, p.Kind, p.R2Key, p.Mime, p.SizeBytes,
 		p.Width, p.Height, p.DurationMs, p.Blurhash, p.FileName,
 		p.URL, p.Title, p.Description, p.SiteName, p.ImageURL,
+		p.EntityType, p.EntityID, p.Snapshot,
 	).Scan(&a.ID, &a.MessageID, &a.Ord, &a.Kind, &a.Mime, &a.SizeBytes, &a.Width,
 		&a.Height, &a.DurationMs, &a.Blurhash, &a.FileName, &a.URL, &a.Title,
-		&a.Description, &a.SiteName, &a.ImageURL, &a.CreatedAt)
+		&a.Description, &a.SiteName, &a.ImageURL,
+		&a.EntityType, &a.EntityID, &a.Entity, &a.CreatedAt)
 	if err != nil {
 		return nil, fmt.Errorf("add attachment: %w", err)
 	}
@@ -583,7 +592,7 @@ func (r *ChatRepo) AttachmentsFor(ctx context.Context, messageIDs []string) (map
 	rows, err := r.pg.Pool.Query(ctx, `
 		SELECT id, message_id, ord, kind, mime, size_bytes, width, height,
 		       duration_ms, blurhash, file_name, url, title, description,
-		       site_name, image_url, created_at
+		       site_name, image_url, entity_type, entity_id, snapshot, created_at
 		FROM message_attachments
 		WHERE message_id = ANY($1)
 		ORDER BY message_id, ord`, messageIDs)
@@ -595,7 +604,8 @@ func (r *ChatRepo) AttachmentsFor(ctx context.Context, messageIDs []string) (map
 		var a model.Attachment
 		if err := rows.Scan(&a.ID, &a.MessageID, &a.Ord, &a.Kind, &a.Mime, &a.SizeBytes,
 			&a.Width, &a.Height, &a.DurationMs, &a.Blurhash, &a.FileName, &a.URL,
-			&a.Title, &a.Description, &a.SiteName, &a.ImageURL, &a.CreatedAt); err != nil {
+			&a.Title, &a.Description, &a.SiteName, &a.ImageURL,
+			&a.EntityType, &a.EntityID, &a.Entity, &a.CreatedAt); err != nil {
 			return nil, fmt.Errorf("scan attachment: %w", err)
 		}
 		out[a.MessageID] = append(out[a.MessageID], a)
@@ -640,7 +650,8 @@ func (r *ChatRepo) Gallery(ctx context.Context, conversationID, userID string, k
 	rows, err := r.pg.Pool.Query(ctx, `
 		SELECT a.id, a.message_id, a.ord, a.kind, a.mime, a.size_bytes, a.width,
 		       a.height, a.duration_ms, a.blurhash, a.file_name, a.url, a.title,
-		       a.description, a.site_name, a.image_url, a.created_at
+		       a.description, a.site_name, a.image_url,
+		       a.entity_type, a.entity_id, a.snapshot, a.created_at
 		FROM message_attachments a
 		JOIN messages m ON m.id = a.message_id
 		WHERE a.conversation_id = $1
@@ -661,7 +672,8 @@ func (r *ChatRepo) Gallery(ctx context.Context, conversationID, userID string, k
 		var a model.Attachment
 		if err := rows.Scan(&a.ID, &a.MessageID, &a.Ord, &a.Kind, &a.Mime, &a.SizeBytes,
 			&a.Width, &a.Height, &a.DurationMs, &a.Blurhash, &a.FileName, &a.URL,
-			&a.Title, &a.Description, &a.SiteName, &a.ImageURL, &a.CreatedAt); err != nil {
+			&a.Title, &a.Description, &a.SiteName, &a.ImageURL,
+			&a.EntityType, &a.EntityID, &a.Entity, &a.CreatedAt); err != nil {
 			return nil, fmt.Errorf("scan gallery row: %w", err)
 		}
 		out = append(out, a)
@@ -1145,6 +1157,60 @@ func (r *ChatRepo) ReportMessage(ctx context.Context, reporterID, messageID, con
 	return nil
 }
 
+// PushCandidate is a participant plus the state that decides whether a new
+// message should reach them out-of-band.
+type PushCandidate struct {
+	UserID      string
+	MutedUntil  *time.Time
+	LastReadSeq int64
+}
+
+// PushCandidates lists everyone still in the thread except the sender.
+func (r *ChatRepo) PushCandidates(ctx context.Context, conversationID, senderID string) ([]PushCandidate, error) {
+	rows, err := r.pg.Pool.Query(ctx, `
+		SELECT user_id, muted_until, last_read_seq
+		FROM conversation_participants
+		WHERE conversation_id = $1 AND user_id <> $2 AND left_at IS NULL`,
+		conversationID, senderID)
+	if err != nil {
+		return nil, fmt.Errorf("push candidates: %w", err)
+	}
+	defer rows.Close()
+
+	out := []PushCandidate{}
+	for rows.Next() {
+		var c PushCandidate
+		if err := rows.Scan(&c.UserID, &c.MutedUntil, &c.LastReadSeq); err != nil {
+			return nil, fmt.Errorf("scan push candidate: %w", err)
+		}
+		out = append(out, c)
+	}
+	return out, rows.Err()
+}
+
+// UnreadConversations counts the caller's threads holding unread messages.
+//
+// Counts THREADS, not messages: the badge answers "is anything waiting for
+// me?", and one chatty group shouldn't dominate it.
+//
+// Muted threads are excluded. Unread still accrues on them and the inbox still
+// shows it — mute silences the alert, and a badge is an alert.
+func (r *ChatRepo) UnreadConversations(ctx context.Context, userID string) (int, error) {
+	var n int
+	err := r.pg.Pool.QueryRow(ctx, `
+		SELECT count(*)
+		FROM conversation_participants p
+		JOIN conversations c ON c.id = p.conversation_id
+		WHERE p.user_id = $1
+		  AND p.left_at IS NULL
+		  AND c.last_seq > p.last_read_seq
+		  AND (p.muted_until IS NULL OR p.muted_until <= now())`, userID).Scan(&n)
+	if err != nil {
+		return 0, fmt.Errorf("count unread conversations: %w", err)
+	}
+	return n, nil
+}
+
 // SetMute sets or clears the caller's mute on a thread. A nil until unmutes.
 //
 // Mute is per participant, not per conversation: muting a group silences it for

internal/repository/post_repo.go

@@ -160,6 +160,46 @@ func (r *PostRepo) FindByID(ctx context.Context, id string) (*model.Post, error)
 	return p, nil
 }
 
+// FindByIDs loads many posts at once, keyed by id. Missing and soft-deleted
+// ids are simply absent from the map.
+//
+// Exists so a page of chat messages sharing N posts costs one query instead of
+// N — the same reason ImagesFor is batched.
+func (r *PostRepo) FindByIDs(ctx context.Context, ids []string) (map[string]*model.Post, error) {
+	out := make(map[string]*model.Post, len(ids))
+	if len(ids) == 0 {
+		return out, nil
+	}
+	q := `SELECT ` + postSelectColumns + postSelectFrom + `WHERE p.id = ANY($1) AND p.deleted_at IS NULL`
+	rows, err := r.pg.Pool.Query(ctx, q, ids)
+	if err != nil {
+		return nil, fmt.Errorf("find posts: %w", err)
+	}
+	defer rows.Close()
+
+	found := make([]string, 0, len(ids))
+	for rows.Next() {
+		p := &model.Post{}
+		if err := scanPost(rows, p); err != nil {
+			return nil, fmt.Errorf("scan post: %w", err)
+		}
+		out[p.ID] = p
+		found = append(found, p.ID)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("find posts: %w", err)
+	}
+
+	imgs, err := r.ImagesFor(ctx, found)
+	if err != nil {
+		return nil, err
+	}
+	for id, p := range out {
+		p.Images = imgs[id]
+	}
+	return out, nil
+}
+
 // ListByHandle returns the profile feed for a user. Unknown handle →
 // empty page (matches pre-existing behaviour).
 func (r *PostRepo) ListByHandle(ctx context.Context, handle string, limit, offset int) ([]model.Post, int, error) {

internal/repository/push_repo.go

@@ -54,6 +54,30 @@ func (r *PushRepo) ListByUser(ctx context.Context, userID string) ([]model.PushS
 
 // PruneToMostRecent keeps only the `keep` most-recently-used subscriptions for
 // a user, deleting older ones — a soft per-user device cap.
+// SubscribedUserIDs lists everyone with at least one live push subscription.
+//
+// Used by broadcasts, which fan out to the whole user base: iterating users
+// and asking each for subscriptions would query for the (large) majority who
+// have never enabled push.
+func (r *PushRepo) SubscribedUserIDs(ctx context.Context) ([]string, error) {
+	rows, err := r.pg.Pool.Query(ctx,
+		`SELECT DISTINCT user_id FROM push_subscriptions`)
+	if err != nil {
+		return nil, fmt.Errorf("subscribed users: %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 subscribed user: %w", err)
+		}
+		out = append(out, id)
+	}
+	return out, rows.Err()
+}
+
 func (r *PushRepo) PruneToMostRecent(ctx context.Context, userID string, keep int) error {
 	if keep < 1 {
 		keep = 1

internal/repository/social_graph_repo.go

@@ -242,6 +242,35 @@ func (r *BlockRepo) Unblock(ctx context.Context, blockerID, blockedID string) er
 
 // BlockState reports, in one round-trip, whether a blocked b and whether
 // b blocked a.
+// BlockedWith returns the subset of others that viewerID is blocked with in
+// EITHER direction — the set whose content must be hidden from them.
+//
+// The batched counterpart to BlockState, for resolving many authors at once.
+func (r *BlockRepo) BlockedWith(ctx context.Context, viewerID string, others []string) (map[string]bool, error) {
+	out := make(map[string]bool, len(others))
+	if viewerID == "" || len(others) == 0 {
+		return out, nil
+	}
+	rows, err := r.pg.Pool.Query(ctx, `
+		SELECT CASE WHEN blocker_id = $1 THEN blocked_id ELSE blocker_id END
+		FROM user_blocks
+		WHERE (blocker_id = $1 AND blocked_id = ANY($2))
+		   OR (blocked_id = $1 AND blocker_id = ANY($2))`,
+		viewerID, others)
+	if err != nil {
+		return nil, fmt.Errorf("blocked with: %w", err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			return nil, fmt.Errorf("scan blocked with: %w", err)
+		}
+		out[id] = true
+	}
+	return out, rows.Err()
+}
+
 func (r *BlockRepo) BlockState(ctx context.Context, a, b string) (aBlockedB, bBlockedA bool, err error) {
 	err = r.pg.Pool.QueryRow(ctx, `
 		SELECT

internal/server/server.go

@@ -77,6 +77,7 @@ type Server struct {
 	feedbackH      *handler.FeedbackHandler
 	statsH         *handler.StatsHandler
 	adminH         *handler.AdminHandler
+	broadcastH     *handler.BroadcastHandler
 
 	// Auth state (nil when Postgres is unavailable — /auth and /users return 503)
 	authIssuer *auth.Issuer
@@ -194,9 +195,6 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) {
 	// across pages (nil-safe — falls back to offset paging without Redis).
 	feedRepo := repository.NewFeedRepo(s.redis, logger)
 	postSvc := service.NewPostService(postRepo, r2, likeRepo, mentionRepo, commentRepo, feedRepo)
-	// Soketi signs webhooks with the app secret; on channel_vacated for a
-	// per-user notifications channel we evict that user's feed seen-set.
-	s.soketiWebhookH = handler.NewSoketiWebhookHandler(cfg.SoketiAppKey, cfg.SoketiAppSecret, feedRepo, logger)
 	bookmarkSvc := service.NewBookmarkService(userRepo, bookmarkRepo)
 	collectionSvc := service.NewCollectionService(userRepo, collectionRepo, postRepo, r2)
 
@@ -259,11 +257,26 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) {
 		cfg.SoketiAppID, cfg.SoketiAppKey, cfg.SoketiAppSecret,
 		cfg.SoketiHost, cfg.SoketiPort, cfg.SoketiUseTLS, logger,
 	)
+	pushRepo := repository.NewPushRepo(s.pg)
+	pushSender := service.NewPushSender(pushRepo, cfg.VAPIDPublicKey, cfg.VAPIDPrivateKey, cfg.VAPIDSubject, logger)
+	s.pushH = handler.NewPushHandler(pushSender, pushRepo)
+	notifPrefsRepo := repository.NewNotificationPrefsRepo(s.pg)
+	notifSvc := service.NewNotificationService(notifRepo, notifPrefsRepo, userRepo, mail, rtClient, pushSender, logger, cfg.FrontendURL)
+
 	// Chat — DMs, groups and the (still coming-soon) Kautilya threads.
 	chatRepo := repository.NewChatRepo(s.pg)
-	chatSvc := service.NewChatService(chatRepo, blockRepo, rtClient, r2, s.redis, logger)
+	chatSvc := service.NewChatService(chatRepo, blockRepo, postRepo,
+		pushSender, notifPrefsRepo, userRepo, rtClient, r2, s.redis, logger)
 	s.chatH = handler.NewChatHandler(chatSvc)
 
+	// Soketi signs webhooks with the app secret. channel_occupied/vacated on a
+	// per-user notifications channel tells us whether they are connected: we
+	// evict the feed seen-set when they leave, and chat uses the same signal
+	// to skip pushing to someone who is already in the app. Constructed after
+	// chatSvc because it depends on it.
+	s.soketiWebhookH = handler.NewSoketiWebhookHandler(
+		cfg.SoketiAppKey, cfg.SoketiAppSecret, feedRepo, chatSvc, logger)
+
 	// Channel authorization for private-* subscriptions. Rules are explicit
 	// and default-deny — an unlisted channel is refused.
 	s.broadcastAuthH = handler.NewBroadcastingAuthHandler(rtClient, logger,
@@ -271,11 +284,6 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) {
 		handler.ConversationRule(chatSvc.IsParticipant),
 	)
 
-	pushRepo := repository.NewPushRepo(s.pg)
-	pushSender := service.NewPushSender(pushRepo, cfg.VAPIDPublicKey, cfg.VAPIDPrivateKey, cfg.VAPIDSubject, logger)
-	s.pushH = handler.NewPushHandler(pushSender, pushRepo)
-	notifPrefsRepo := repository.NewNotificationPrefsRepo(s.pg)
-	notifSvc := service.NewNotificationService(notifRepo, notifPrefsRepo, userRepo, mail, rtClient, pushSender, logger, cfg.FrontendURL)
 	// Wire the notification seam into the post service so a new like notifies
 	// the author (push + realtime; not the liker themselves).
 	postSvc.SetNotifier(notifSvc)
@@ -336,6 +344,14 @@ func (s *Server) initAuth(cfg *config.Config, logger *zap.Logger) {
 	adminRepo := repository.NewAdminRepo(s.pg)
 	adminSvc := service.NewAdminService(adminRepo, statsRepo, notifSvc, feedbackSvc, userRepo, s.discipleRepo, mail, logger, cfg.FrontendURL)
 	s.adminH = handler.NewAdminHandler(adminSvc)
+
+	// Broadcasts — admin announcements shown once per user. No notification
+	// rows: the modal is the surface, and push plus the realtime nudge only
+	// tell clients to go and check for one.
+	broadcastSvc := service.NewBroadcastService(
+		repository.NewBroadcastRepo(s.pg), pushSender, pushRepo,
+		notifPrefsRepo, rtClient, r2, logger)
+	s.broadcastH = handler.NewBroadcastHandler(broadcastSvc)
 }
 
 // initShortener wires Postgres + Redis + the shorturl handler/workers.
@@ -905,10 +921,20 @@ func (s *Server) setupRouter() {
 		// Chat — inbox, threads, messages. Every route requires a session;
 		// membership is re-checked per request in the service, so a
 		// conversation id on its own grants nothing.
+		if s.broadcastH != nil {
+			// Announcements: fetch the one waiting, and dismiss it.
+			r.Route("/broadcasts", func(r chi.Router) {
+				r.Use(middleware.RequireAuth(s.authIssuer, s.userRepo))
+				r.Get("/pending", s.broadcastH.Pending)
+				r.Post("/{id}/seen", s.broadcastH.Seen)
+			})
+		}
+
 		if s.chatH != nil {
 			r.Route("/chat", func(r chi.Router) {
 				r.Use(middleware.RequireAuth(s.authIssuer, s.userRepo))
 				r.Get("/conversations", s.chatH.Inbox)
+				r.Get("/unread", s.chatH.Unread)
 				r.Post("/conversations/dm", s.chatH.OpenDM)
 				r.Post("/conversations/group", s.chatH.CreateGroup)
 				r.Post("/conversations/kautilya", s.chatH.OpenAIThread)
@@ -1125,6 +1151,13 @@ func (s *Server) setupRouter() {
 			// Direct-disciple linking: mark an account as one of Śrīla
 			// Prabhupāda's initiated disciples (or clear the mark).
 			r.Get("/disciples", s.adminH.ListDisciples)
+
+			// Announcements composed and published from the admin panel.
+			r.Get("/broadcasts", s.broadcastH.List)
+			r.Post("/broadcasts", s.broadcastH.Create)
+			r.Post("/broadcasts/banner-presign", s.broadcastH.PresignBanner)
+			r.Post("/broadcasts/{id}/publish", s.broadcastH.Publish)
+			r.Delete("/broadcasts/{id}", s.broadcastH.Delete)
 			r.Post("/users/{id}/disciple", s.adminH.LinkDisciple)
 			r.Delete("/users/{id}/disciple", s.adminH.UnlinkDisciple)
 			r.Get("/posts", s.adminH.ListPosts)

internal/service/broadcast_service.go

@@ -0,0 +1,266 @@
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/url"
+	"strings"
+	"time"
+
+	"github.com/prabhupadasays/spsays-bff/internal/model"
+	"github.com/prabhupadasays/spsays-bff/internal/repository"
+	"github.com/prabhupadasays/spsays-bff/pkg/realtime"
+	"go.uber.org/zap"
+)
+
+var (
+	ErrBroadcastTitle  = errors.New("a title is required")
+	ErrBroadcastBody   = errors.New("a description is required")
+	ErrBroadcastCTA    = errors.New("that CTA link is not a valid URL")
+	ErrBroadcastBanner = errors.New("that image was not uploaded for a broadcast")
+)
+
+const (
+	maxBroadcastTitleRunes = 200
+	maxBroadcastBodyRunes  = 4000
+	// bannerPrefix namespaces broadcast banners in the PUBLIC bucket. An
+	// announcement is not private, and a signed URL would expire while the
+	// modal sat open on someone's screen.
+	bannerPrefix = "broadcasts/"
+)
+
+// BroadcastChannel is the PUBLIC realtime channel every client watches for
+// announcements.
+//
+// Public is deliberate and safe because the event carries no content — just
+// "something was published". Each client then fetches /broadcasts/pending,
+// which is authenticated and per-user, so someone who has already dismissed it
+// gets nothing. That also keeps publishing to one Soketi call instead of one
+// per user.
+const BroadcastChannel = "broadcasts"
+
+type BroadcastService struct {
+	repo   *repository.BroadcastRepo
+	push   *PushSender
+	pushes *repository.PushRepo
+	prefs  *repository.NotificationPrefsRepo
+	rt     *realtime.SoketiClient
+	r2     *repository.R2
+	logger *zap.Logger
+}
+
+func NewBroadcastService(
+	repo *repository.BroadcastRepo,
+	push *PushSender,
+	pushes *repository.PushRepo,
+	prefs *repository.NotificationPrefsRepo,
+	rt *realtime.SoketiClient,
+	r2 *repository.R2,
+	logger *zap.Logger,
+) *BroadcastService {
+	return &BroadcastService{
+		repo: repo, push: push, pushes: pushes, prefs: prefs,
+		rt: rt, r2: r2, logger: logger,
+	}
+}
+
+// CreateParams is an admin-authored announcement.
+type CreateParams struct {
+	Title     string
+	Body      string
+	BannerKey *string
+	CTALabel  *string
+	CTAURL    *string
+	ExpiresAt *time.Time
+}
+
+// Create stores a draft. Nothing is delivered until Publish.
+func (s *BroadcastService) Create(ctx context.Context, adminID string, p CreateParams) (*model.Broadcast, error) {
+	title := strings.TrimSpace(p.Title)
+	body := strings.TrimSpace(p.Body)
+	if title == "" || len([]rune(title)) > maxBroadcastTitleRunes {
+		return nil, ErrBroadcastTitle
+	}
+	if body == "" || len([]rune(body)) > maxBroadcastBodyRunes {
+		return nil, ErrBroadcastBody
+	}
+	if p.CTAURL != nil && strings.TrimSpace(*p.CTAURL) != "" {
+		clean, ok := safeCTAURL(*p.CTAURL)
+		if !ok {
+			return nil, ErrBroadcastCTA
+		}
+		p.CTAURL = &clean
+	} else {
+		p.CTAURL = nil
+	}
+	if p.BannerKey != nil && !strings.HasPrefix(*p.BannerKey, bannerPrefix) {
+		return nil, ErrBroadcastBanner
+	}
+
+	// Same parser as the chat composer, so the admin writes *bold* / ~strike~ /
+	// `code` and links are detected — and no HTML is ever stored.
+	clean, entities := ParseEntities(body)
+	encoded, err := json.Marshal(entities)
+	if err != nil {
+		return nil, fmt.Errorf("encode entities: %w", err)
+	}
+	if entities == nil {
+		encoded = []byte(`[]`)
+	}
+
+	b, key, err := s.repo.Create(ctx, repository.CreateBroadcastParams{
+		Title: title, Body: clean, Entities: encoded, BannerKey: p.BannerKey,
+		CTALabel: p.CTALabel, CTAURL: p.CTAURL, ExpiresAt: p.ExpiresAt,
+		CreatedBy: adminID,
+	})
+	if err != nil {
+		return nil, err
+	}
+	s.withBanner(b, key)
+	return b, nil
+}
+
+// Publish makes a broadcast live, then nudges everyone: connected clients over
+// realtime, absent ones by push.
+func (s *BroadcastService) Publish(ctx context.Context, id string) (*model.Broadcast, error) {
+	b, key, err := s.repo.Publish(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	s.withBanner(b, key)
+
+	// One trigger for the whole user base. Content-free on purpose — clients
+	// re-fetch their own pending state.
+	if s.rt != nil {
+		if err := s.rt.Trigger(ctx, BroadcastChannel, "broadcast_published", map[string]any{}); err != nil {
+			s.logger.Warn("broadcast: realtime nudge failed", zap.Error(err))
+		}
+	}
+	go s.fanOutPush(context.WithoutCancel(ctx), b)
+	return b, nil
+}
+
+// fanOutPush notifies everyone with a push subscription who hasn't opted out
+// of announcements.
+//
+// Runs detached from the request: a large user base takes far longer than an
+// admin should wait on a button.
+func (s *BroadcastService) fanOutPush(ctx context.Context, b *model.Broadcast) {
+	if s.push == nil || !s.push.Enabled() || s.pushes == nil {
+		return
+	}
+	users, err := s.pushes.SubscribedUserIDs(ctx)
+	if err != nil {
+		s.logger.Warn("broadcast: listing subscribers failed", zap.Error(err))
+		return
+	}
+
+	sent := 0
+	for _, uid := range users {
+		if s.prefs != nil {
+			// Reuses the existing "Community" category, described in settings
+			// as rare announcements — which is exactly this. Someone who
+			// opted out of those stays opted out.
+			if off, err := s.prefs.DisabledSet(ctx, uid); err == nil &&
+				prefOff(off, NotifDiscipleJoined, repository.ChannelPush) {
+				continue
+			}
+		}
+		s.push.SendToUser(ctx, uid, PushPayload{
+			Title: b.Title,
+			Body:  truncateRunes(b.Body, 160),
+			URL:   "/",
+			// One tag per broadcast: a re-publish replaces rather than stacks.
+			Tag: "broadcast:" + b.ID,
+		})
+		sent++
+	}
+	s.logger.Info("broadcast: push fan-out complete",
+		zap.String("broadcast_id", b.ID), zap.Int("recipients", sent))
+}
+
+// Pending returns the announcement waiting for this user, or nil.
+func (s *BroadcastService) Pending(ctx context.Context, userID string) (*model.Broadcast, error) {
+	b, key, err := s.repo.Pending(ctx, userID)
+	if err != nil || b == nil {
+		return nil, err
+	}
+	s.withBanner(b, key)
+	return b, nil
+}
+
+// MarkSeen records that a user dismissed it — the "at most once" half.
+func (s *BroadcastService) MarkSeen(ctx context.Context, id, userID string) error {
+	return s.repo.MarkSeen(ctx, id, userID)
+}
+
+func (s *BroadcastService) List(ctx context.Context, limit int) ([]model.Broadcast, error) {
+	if limit <= 0 || limit > 100 {
+		limit = 50
+	}
+	items, keys, err := s.repo.List(ctx, limit)
+	if err != nil {
+		return nil, err
+	}
+	for i := range items {
+		s.withBanner(&items[i], keys[i])
+	}
+	return items, nil
+}
+
+func (s *BroadcastService) Delete(ctx context.Context, id string) error {
+	return s.repo.Delete(ctx, id)
+}
+
+// PresignBanner issues an upload URL for a banner image.
+func (s *BroadcastService) PresignBanner(ctx context.Context, contentType string) (uploadURL, key string, err error) {
+	if s.r2 == nil {
+		return "", "", errors.New("media uploads are not configured")
+	}
+	kind, ok := attachmentKindFor(contentType)
+	if !ok || kind != model.AttachmentImage {
+		return "", "", ErrChatBadMedia
+	}
+	return presignImage(ctx, s.r2, strings.TrimSuffix(bannerPrefix, "/"), "banner", contentType)
+}
+
+func (s *BroadcastService) withBanner(b *model.Broadcast, key *string) {
+	if key == nil || *key == "" || s.r2 == nil {
+		return
+	}
+	url := s.r2.PublicURL(*key)
+	b.BannerURL = &url
+}
+
+// safeCTAURL accepts an in-app path or an absolute http(s) URL.
+//
+// Admin-authored, so this is a typo guard more than a threat model — but
+// `javascript:` and `data:` still have no business in a link the whole user
+// base is invited to tap.
+func safeCTAURL(raw string) (string, bool) {
+	raw = strings.TrimSpace(raw)
+	if raw == "" || len(raw) > 2048 {
+		return "", false
+	}
+	if strings.ContainsAny(raw, "\\") || strings.ContainsAny(raw, "\n\r\t") {
+		return "", false
+	}
+	if strings.HasPrefix(raw, "/") {
+		// Protocol-relative "//evil.example" is not an in-app path.
+		if strings.HasPrefix(raw, "//") {
+			return "", false
+		}
+		u, err := url.Parse(raw)
+		if err != nil || u.Scheme != "" || u.Host != "" {
+			return "", false
+		}
+		return raw, true
+	}
+	u, err := url.Parse(raw)
+	if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
+		return "", false
+	}
+	return u.String(), true
+}

internal/service/broadcast_service_test.go

@@ -0,0 +1,166 @@
+package service
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/joho/godotenv"
+	"github.com/prabhupadasays/spsays-bff/internal/repository"
+	"go.uber.org/zap"
+)
+
+func testBroadcastService(t *testing.T) (*BroadcastService, *repository.Postgres) {
+	t.Helper()
+	uri := os.Getenv("POSTGRES_URI")
+	if uri == "" {
+		_ = godotenv.Load(filepath.Join("..", "..", ".env"))
+		uri = os.Getenv("POSTGRES_URI")
+	}
+	if uri == "" {
+		t.Skip("POSTGRES_URI not set")
+	}
+	pg, err := repository.NewPostgres(context.Background(), uri, zap.NewNop())
+	if err != nil {
+		t.Skipf("postgres unreachable: %v", err)
+	}
+	t.Cleanup(pg.Close)
+	// Nil push/realtime/R2: these tests are about who sees what, and none of
+	// those services are reachable from a test run.
+	return NewBroadcastService(repository.NewBroadcastRepo(pg), nil, nil, nil, nil, nil, zap.NewNop()), pg
+}
+
+func bcUser(t *testing.T, pg *repository.Postgres, label string) string {
+	t.Helper()
+	id := uuid.NewString()
+	suffix := id[:8]
+	if _, err := pg.Pool.Exec(context.Background(), `
+		INSERT INTO users (id, email, password_hash, name, handle)
+		VALUES ($1, $2, 'x', $3, $4)`,
+		id, "bc-"+suffix+"@example.invalid", "BC "+label, "bc_"+suffix); err != nil {
+		t.Fatalf("create user: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pg.Pool.Exec(context.Background(), `DELETE FROM users WHERE id = $1`, id)
+	})
+	return id
+}
+
+func cleanupBroadcast(t *testing.T, pg *repository.Postgres, id string) {
+	t.Cleanup(func() {
+		_, _ = pg.Pool.Exec(context.Background(), `DELETE FROM broadcasts WHERE id = $1`, id)
+	})
+}
+
+// "At least and at most once" is the whole contract. A user must be offered a
+// published announcement until they dismiss it, and never again after.
+func TestBroadcast_ShownExactlyOncePerUser(t *testing.T) {
+	svc, pg := testBroadcastService(t)
+	ctx := context.Background()
+	admin, alice, bob := bcUser(t, pg, "Admin"), bcUser(t, pg, "A"), bcUser(t, pg, "B")
+
+	b, err := svc.Create(ctx, admin, CreateParams{
+		Title: "New: chat", Body: "Direct messages are here.",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	cleanupBroadcast(t, pg, b.ID)
+
+	// A draft is nobody's business yet.
+	if got, _ := svc.Pending(ctx, alice); got != nil {
+		t.Fatalf("an unpublished draft was offered: %+v", got)
+	}
+
+	if _, err := svc.Publish(ctx, b.ID); err != nil {
+		t.Fatal(err)
+	}
+
+	// At least once: offered, and still offered until dismissed.
+	for i := 0; i < 2; i++ {
+		got, err := svc.Pending(ctx, alice)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if got == nil || got.ID != b.ID {
+			t.Fatalf("call %d: expected the published announcement, got %+v", i+1, got)
+		}
+	}
+
+	// At most once: dismissed means gone, and dismissing twice is harmless.
+	if err := svc.MarkSeen(ctx, b.ID, alice); err != nil {
+		t.Fatal(err)
+	}
+	if err := svc.MarkSeen(ctx, b.ID, alice); err != nil {
+		t.Errorf("second dismissal should be idempotent, got %v", err)
+	}
+	if got, _ := svc.Pending(ctx, alice); got != nil {
+		t.Errorf("shown again after dismissal: %+v", got)
+	}
+
+	// One person dismissing must not dismiss it for everyone.
+	if got, _ := svc.Pending(ctx, bob); got == nil || got.ID != b.ID {
+		t.Errorf("another user lost the announcement: %+v", got)
+	}
+}
+
+// An announcement past its expiry stops being offered, so someone signing up
+// months later isn't shown last quarter's news.
+func TestBroadcast_ExpiryStopsDelivery(t *testing.T) {
+	svc, pg := testBroadcastService(t)
+	ctx := context.Background()
+	admin, reader := bcUser(t, pg, "Admin"), bcUser(t, pg, "R")
+
+	past := time.Now().Add(-time.Hour)
+	b, err := svc.Create(ctx, admin, CreateParams{
+		Title: "Old news", Body: "Yesterday's announcement.", ExpiresAt: &past,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	cleanupBroadcast(t, pg, b.ID)
+	if _, err := svc.Publish(ctx, b.ID); err != nil {
+		t.Fatal(err)
+	}
+
+	if got, _ := svc.Pending(ctx, reader); got != nil {
+		t.Errorf("an expired announcement was offered: %+v", got)
+	}
+}
+
+// Only the newest announcement is offered: two modals stacked on launch is a
+// worse experience than missing the older one.
+func TestBroadcast_OffersNewestFirst(t *testing.T) {
+	svc, pg := testBroadcastService(t)
+	ctx := context.Background()
+	admin, reader := bcUser(t, pg, "Admin"), bcUser(t, pg, "R")
+
+	older, err := svc.Create(ctx, admin, CreateParams{Title: "First", Body: "one"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	cleanupBroadcast(t, pg, older.ID)
+	if _, err := svc.Publish(ctx, older.ID); err != nil {
+		t.Fatal(err)
+	}
+
+	newer, err := svc.Create(ctx, admin, CreateParams{Title: "Second", Body: "two"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	cleanupBroadcast(t, pg, newer.ID)
+	if _, err := svc.Publish(ctx, newer.ID); err != nil {
+		t.Fatal(err)
+	}
+
+	got, err := svc.Pending(ctx, reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got == nil || got.ID != newer.ID {
+		t.Errorf("expected the newest announcement, got %+v", got)
+	}
+}

internal/service/chat_service.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"net/url"
 	"path"
 	"strings"
 	"time"
@@ -41,6 +42,8 @@ var (
 	ErrAvatarUnset          = errors.New("this group has no photo")
 	ErrChatNotAccepting     = errors.New("this person isn't accepting messages from you")
 	ErrChatBadWhoCanMessage = errors.New("unknown messaging restriction")
+	ErrChatBadQuote         = errors.New("that quote is not a valid passage reference")
+	ErrChatBadEntity        = errors.New("that post cannot be shared")
 	ErrChatBadAccountAge    = errors.New("account age must be between 1 and 120 months")
 	ErrChatGenderUnset      = errors.New("add your gender in profile settings to use this option")
 )
@@ -60,6 +63,14 @@ const (
 type ChatService struct {
 	chat   *repository.ChatRepo
 	blocks *repository.BlockRepo
+	// posts resolves shared post cards. Optional: nil leaves entity
+	// attachments unresolved rather than failing a whole thread load.
+	posts *repository.PostRepo
+	// push, prefs and users deliver out-of-band message alerts. All optional:
+	// nil means no push, never a failed send.
+	push   *PushSender
+	prefs  *repository.NotificationPrefsRepo
+	users  *repository.UserRepo
 	rt     *realtime.SoketiClient
 	r2     *repository.R2
 	links  *unfurl.Fetcher
@@ -70,13 +81,18 @@ type ChatService struct {
 func NewChatService(
 	chat *repository.ChatRepo,
 	blocks *repository.BlockRepo,
+	posts *repository.PostRepo,
+	push *PushSender,
+	prefs *repository.NotificationPrefsRepo,
+	users *repository.UserRepo,
 	rt *realtime.SoketiClient,
 	r2 *repository.R2,
 	cache *repository.Redis,
 	logger *zap.Logger,
 ) *ChatService {
 	return &ChatService{
-		chat: chat, blocks: blocks, rt: rt, r2: r2,
+		chat: chat, blocks: blocks, posts: posts,
+		push: push, prefs: prefs, users: users, rt: rt, r2: r2,
 		links: unfurl.New(), cache: cache, logger: logger,
 	}
 }
@@ -149,6 +165,11 @@ func (s *ChatService) Inbox(ctx context.Context, userID, tab string, limit int)
 	return s.chat.Inbox(ctx, userID, KindsForTab(tab), limit)
 }
 
+// UnreadCount reports how many threads are waiting on the caller.
+func (s *ChatService) UnreadCount(ctx context.Context, userID string) (int, error) {
+	return s.chat.UnreadConversations(ctx, userID)
+}
+
 // OpenDM returns (creating if needed) the DM between the caller and peer.
 //
 // Blocking is enforced in BOTH directions: a blocked user must not be able to
@@ -315,11 +336,12 @@ func (s *ChatService) Starred(ctx context.Context, userID, conversationID string
 	if err != nil {
 		return nil, err
 	}
-	return s.hydrateAttachments(ctx, msgs)
+	return s.hydrateAttachments(ctx, userID, msgs)
 }
 
-// hydrateAttachments fills in each message's attachments in one query.
+// hydrateAttachments fills in each message's attachments in one query, then
-func (s *ChatService) hydrateAttachments(ctx context.Context, msgs []model.Message) ([]model.Message, error) {
+// resolves any shared entities for this particular viewer.
+func (s *ChatService) hydrateAttachments(ctx context.Context, viewerID string, msgs []model.Message) ([]model.Message, error) {
 	ids := make([]string, 0, len(msgs))
 	for i := range msgs {
 		ids = append(ids, msgs[i].ID)
@@ -336,6 +358,11 @@ func (s *ChatService) hydrateAttachments(ctx context.Context, msgs []model.Messa
 			msgs[i].Attachments = []model.Attachment{}
 		}
 	}
+	pages := make([][]model.Attachment, 0, len(msgs))
+	for i := range msgs {
+		pages = append(pages, msgs[i].Attachments)
+	}
+	s.resolveEntities(ctx, viewerID, pages)
 	return msgs, nil
 }
 
@@ -362,7 +389,7 @@ func (s *ChatService) Messages(ctx context.Context, userID, conversationID strin
 		return nil, err
 	}
 
-	return s.hydrateAttachments(ctx, msgs)
+	return s.hydrateAttachments(ctx, userID, msgs)
 }
 
 // SendRequest is one outgoing message from a client.
@@ -377,6 +404,11 @@ type SendRequest struct {
 	// fetches the title, description and image itself, because a client that
 	// could supply them could make any link claim to be anything.
 	PreviewURL *string
+	// Entity is something else in the app — a post — shared as a card.
+	Entity *EntityAttachment
+	// Quote is a passage shared from the library, rendered as a quote the
+	// reader can tap to return to the exact phrase.
+	Quote *QuoteAttachment
 	// Attachments are files already uploaded to the private bucket via
 	// PresignAttachment; their keys are checked against the caller's prefix.
 	Attachments []MediaAttachment
@@ -407,10 +439,14 @@ func (s *ChatService) Send(ctx context.Context, userID, conversationID string, r
 		return nil, err
 	}
 
+	if req.Quote != nil && !req.Quote.valid() {
+		return nil, ErrChatBadQuote
+	}
+
 	body := strings.TrimSpace(req.Body)
-	// A photo with no caption is an ordinary message; only a message with
+	// A photo or a shared passage with no caption is an ordinary message; only
-	// neither text nor attachments is empty.
+	// a message with nothing at all in it is empty.
-	if body == "" && len(req.Attachments) == 0 {
+	if body == "" && len(req.Attachments) == 0 && req.Quote == nil && req.Entity == nil {
 		return nil, ErrChatEmptyBody
 	}
 	if len([]rune(body)) > maxBodyRunes {
@@ -442,13 +478,17 @@ func (s *ChatService) Send(ctx context.Context, userID, conversationID string, r
 	// the first time, and re-publishing would duplicate the bubble for
 	// everyone else.
 	if !created {
-		return s.withAttachments(ctx, msg)
+		return s.withAttachments(ctx, userID, msg)
 	}
 
+	if err := s.attachEntity(ctx, conversationID, userID, msg, req.Entity); err != nil {
+		s.logger.Warn("chat: entity share refused", zap.String("user_id", userID), zap.Error(err))
+	}
+	s.attachQuote(ctx, conversationID, msg, req.Quote)
 	s.attachMedia(ctx, conversationID, userID, msg, req.Attachments)
 	s.attachLinkPreview(ctx, conversationID, msg, entities, req.PreviewURL)
 
-	full, err := s.withAttachments(ctx, msg)
+	full, err := s.withAttachments(ctx, userID, msg)
 	if err != nil {
 		full = msg
 	}
@@ -457,7 +497,7 @@ func (s *ChatService) Send(ctx context.Context, userID, conversationID string, r
 }
 
 // withAttachments hydrates a message's attachments for the response.
-func (s *ChatService) withAttachments(ctx context.Context, msg *model.Message) (*model.Message, error) {
+func (s *ChatService) withAttachments(ctx context.Context, viewerID string, msg *model.Message) (*model.Message, error) {
 	byMessage, err := s.chat.AttachmentsFor(ctx, []string{msg.ID})
 	if err != nil {
 		return msg, err
@@ -466,6 +506,7 @@ func (s *ChatService) withAttachments(ctx context.Context, msg *model.Message) (
 	if msg.Attachments == nil {
 		msg.Attachments = []model.Attachment{}
 	}
+	s.resolveEntities(ctx, viewerID, [][]model.Attachment{msg.Attachments})
 	return msg, nil
 }
 
@@ -499,6 +540,262 @@ func (s *ChatService) attachMedia(ctx context.Context, conversationID, userID st
 	}
 }
 
+// EntityAttachment is a reference to something else in the app, shared into a
+// chat. Only posts for now; the column allows comments, profiles and
+// collections when their cards are built.
+type EntityAttachment struct {
+	Type string
+	ID   string
+}
+
+// postSnapshot is the copy stored alongside the reference.
+//
+// It is a FALLBACK, never the source of truth: resolveEntities re-reads the
+// live post for each viewer, because a post can be deleted, hidden, or become
+// invisible to one participant after it was shared. Rendering the snapshot
+// unconditionally would leak a post to someone who has since lost access.
+type postSnapshot struct {
+	ID              string  `json:"id"`
+	AuthorHandle    string  `json:"author_handle"`
+	AuthorName      string  `json:"author_name"`
+	AuthorAvatarURL *string `json:"author_avatar_url,omitempty"`
+	Caption         string  `json:"caption"`
+	ImageURL        *string `json:"image_url,omitempty"`
+	SourceTitle     string  `json:"source_title,omitempty"`
+}
+
+func snapshotOf(p *model.Post) postSnapshot {
+	snap := postSnapshot{
+		ID: p.ID, AuthorHandle: p.AuthorHandle, AuthorName: p.AuthorName,
+		AuthorAvatarURL: p.AuthorAvatarURL, Caption: p.Caption,
+		SourceTitle: p.SourceTitle,
+	}
+	if len(p.Images) > 0 {
+		url := p.Images[0].URL
+		snap.ImageURL = &url
+	}
+	return snap
+}
+
+// postVisibleTo decides whether a viewer may see a post inside a chat bubble.
+//
+// Same rules the feed applies: a hidden post is the owner's alone, and a block
+// in either direction hides the author's content.
+func (s *ChatService) postVisibleTo(ctx context.Context, viewerID string, p *model.Post) bool {
+	if p.IsHidden && p.UserID != viewerID {
+		return false
+	}
+	if p.UserID == viewerID {
+		return true
+	}
+	blocked, blockedBy, err := s.blocks.BlockState(ctx, viewerID, p.UserID)
+	if err != nil {
+		// Fail closed: an unreadable block table must not open a post up.
+		return false
+	}
+	return !blocked && !blockedBy
+}
+
+// attachEntity stores a shared post after checking the SENDER can see it —
+// you cannot forward what you were never allowed to read.
+func (s *ChatService) attachEntity(ctx context.Context, conversationID, userID string, msg *model.Message, e *EntityAttachment) error {
+	if e == nil {
+		return nil
+	}
+	if e.Type != model.EntityPost {
+		return ErrChatBadEntity
+	}
+	post, err := s.posts.FindByID(ctx, e.ID)
+	if err != nil || post == nil {
+		return ErrChatBadEntity
+	}
+	if !s.postVisibleTo(ctx, userID, post) {
+		return ErrChatBadEntity
+	}
+
+	snap, err := json.Marshal(snapshotOf(post))
+	if err != nil {
+		return ErrChatBadEntity
+	}
+	entityType, entityID := model.EntityPost, post.ID
+	if _, err := s.chat.AddAttachment(ctx, repository.AddAttachmentParams{
+		MessageID: msg.ID, ConversationID: conversationID, Ord: 0,
+		Kind: model.AttachmentEntity, EntityType: &entityType,
+		EntityID: &entityID, Snapshot: snap,
+	}); err != nil {
+		s.logger.Warn("chat: attach entity failed", zap.Error(err))
+	}
+	return nil
+}
+
+// resolveEntities re-reads every shared entity on a page for THIS viewer.
+//
+// This is the permission re-check that makes sharing safe: access is decided
+// at read time, per person, not frozen at send time. A post that was visible
+// when shared and is hidden now renders as unavailable.
+//
+// It takes the whole page at once — three queries (posts, their images,
+// blocks) regardless of how many shares are on it. Resolving per message
+// instead would cost two queries per share and could not dedupe a post quoted
+// in several messages.
+func (s *ChatService) resolveEntities(ctx context.Context, viewerID string, pages [][]model.Attachment) {
+	if s.posts == nil {
+		return
+	}
+
+	// Gather the distinct posts referenced anywhere on the page.
+	shared := map[string]bool{}
+	for _, atts := range pages {
+		for i := range atts {
+			if atts[i].Kind == model.AttachmentEntity && atts[i].EntityID != nil {
+				shared[*atts[i].EntityID] = true
+			}
+		}
+	}
+	if len(shared) == 0 {
+		return
+	}
+	ids := make([]string, 0, len(shared))
+	for id := range shared {
+		ids = append(ids, id)
+	}
+
+	posts, err := s.posts.FindByIDs(ctx, ids)
+	if err != nil {
+		// Fail closed: every card becomes a tombstone rather than falling back
+		// to snapshots whose visibility we could not confirm.
+		s.logger.Warn("chat: resolving shared posts failed", zap.Error(err))
+		posts = nil
+	}
+
+	authors := make([]string, 0, len(posts))
+	for _, post := range posts {
+		if post.UserID != viewerID {
+			authors = append(authors, post.UserID)
+		}
+	}
+	blocked, err := s.blocks.BlockedWith(ctx, viewerID, authors)
+	if err != nil {
+		s.logger.Warn("chat: block lookup for shared posts failed", zap.Error(err))
+		// Fail closed again: an unreadable block table must not open posts up.
+		blocked = nil
+		for _, id := range authors {
+			if blocked == nil {
+				blocked = map[string]bool{}
+			}
+			blocked[id] = true
+		}
+	}
+
+	// Resolve each referenced post once, then fan the result out.
+	resolved := make(map[string]json.RawMessage, len(shared))
+	for id := range shared {
+		post := posts[id]
+		if post == nil {
+			continue
+		}
+		if post.IsHidden && post.UserID != viewerID {
+			continue
+		}
+		if blocked[post.UserID] {
+			continue
+		}
+		if b, mErr := json.Marshal(snapshotOf(post)); mErr == nil {
+			resolved[id] = b
+		}
+	}
+
+	for _, atts := range pages {
+		for i := range atts {
+			a := &atts[i]
+			if a.Kind != model.AttachmentEntity || a.EntityID == nil {
+				continue
+			}
+			// The stored snapshot is never sent to a client — only a copy
+			// resolved for this viewer — so a stale one cannot leak.
+			a.Entity = resolved[*a.EntityID]
+			a.EntityUnavailable = a.Entity == nil
+		}
+	}
+}
+
+// QuoteAttachment is a passage shared out of the library.
+type QuoteAttachment struct {
+	// Path is an APP-RELATIVE deep link, e.g. "/books/bg/2/13?hl=…".
+	//
+	// Deliberately not a full URL. A quote renders as trusted in-app content,
+	// so allowing an absolute URL would let a sender dress an off-site link up
+	// as a scripture citation. A path cannot leave the app, which removes the
+	// question of which origins to trust.
+	Path string
+	// Title is the source reference shown above the passage.
+	Title string
+	// Text is the quoted passage itself.
+	Text string
+}
+
+const (
+	maxQuoteTitleRunes = 200
+	maxQuoteTextRunes  = 2000
+	maxQuotePathBytes  = 2048
+)
+
+func (q *QuoteAttachment) valid() bool {
+	if strings.TrimSpace(q.Text) == "" {
+		return false
+	}
+	if len([]rune(q.Title)) > maxQuoteTitleRunes || len([]rune(q.Text)) > maxQuoteTextRunes {
+		return false
+	}
+	return safeAppPath(q.Path)
+}
+
+// safeAppPath accepts only a path within this app.
+//
+// "//evil.com" and "/\evil.com" are protocol-relative and would navigate
+// off-site, so a leading slash alone is not enough to check.
+func safeAppPath(p string) bool {
+	if p == "" || len(p) > maxQuotePathBytes {
+		return false
+	}
+	if p[0] != '/' {
+		return false
+	}
+	if strings.HasPrefix(p, "//") || strings.HasPrefix(p, "/\\") {
+		return false
+	}
+	if strings.ContainsAny(p, "\\") {
+		return false
+	}
+	for _, r := range p {
+		if r < 0x20 || r == 0x7f {
+			return false
+		}
+	}
+	// A parsed path must not carry a scheme or host of its own.
+	u, err := url.Parse(p)
+	if err != nil || u.Scheme != "" || u.Host != "" {
+		return false
+	}
+	return true
+}
+
+// attachQuote stores a shared passage. Unlike a link card, the contents come
+// from the client — but they are only ever rendered as a quotation, and the
+// path has already been constrained to this app.
+func (s *ChatService) attachQuote(ctx context.Context, conversationID string, msg *model.Message, q *QuoteAttachment) {
+	if q == nil {
+		return
+	}
+	path, title, text := q.Path, strings.TrimSpace(q.Title), strings.TrimSpace(q.Text)
+	if _, err := s.chat.AddAttachment(ctx, repository.AddAttachmentParams{
+		MessageID: msg.ID, ConversationID: conversationID, Ord: 0,
+		Kind: model.AttachmentQuote, URL: &path, Title: &title, Description: &text,
+	}); err != nil {
+		s.logger.Warn("chat: attach quote failed", zap.Error(err))
+	}
+}
+
 // attachLinkPreview unfurls the chosen link and stores the card.
 //
 // Two rules make this safe. The URL must be one the parser actually found in
@@ -736,7 +1033,7 @@ func (s *ChatService) ForwardMessage(ctx context.Context, userID, messageID, toC
 		return nil, err
 	}
 	if created {
-		full, hErr := s.withAttachments(ctx, msg)
+		full, hErr := s.withAttachments(ctx, userID, msg)
 		if hErr == nil {
 			msg = full
 		}
@@ -821,6 +1118,14 @@ func (s *ChatService) systemMessage(ctx context.Context, conversationID, body st
 		return
 	}
 	if created {
+		// Publish the hydrated message. Append leaves Attachments nil, which
+		// serialises to null and breaks clients that trust the "always an
+		// array" contract this field's own doc comment promises.
+		if full, hErr := s.withAttachments(ctx, "", msg); hErr == nil {
+			msg = full
+		} else if msg.Attachments == nil {
+			msg.Attachments = []model.Attachment{}
+		}
 		s.publish(ctx, conversationID, "", msg)
 	}
 }
@@ -1296,6 +1601,13 @@ func (s *ChatService) Receipts(ctx context.Context, userID, conversationID strin
 // publish fans a new message out to the open thread, and separately nudges each
 // other participant's inbox.
 func (s *ChatService) publish(ctx context.Context, conversationID, senderID string, msg *model.Message) {
+	// Out-of-band alerts for anyone not currently connected. Best-effort and
+	// after the realtime fan-out, so a slow push never delays delivery to
+	// people who are online.
+	if conv, err := s.chat.Conversation(ctx, conversationID); err == nil {
+		s.notifyNewMessage(ctx, conv, senderID, msg)
+	}
+
 	if s.rt == nil {
 		return
 	}
@@ -1330,6 +1642,149 @@ func (s *ChatService) publish(ctx context.Context, conversationID, senderID stri
 	}
 }
 
+// pushConfig tunes out-of-band delivery.
+const (
+	// One push per thread per window, however many messages arrive. A burst
+	// should buzz a pocket once.
+	pushDebounceWindow = 90 * time.Second
+	// How long a live WebSocket connection suppresses push for.
+	presenceTTL = 2 * time.Minute
+	// Lock-screen previews are deliberately senderless of content — see
+	// notifyNewMessage.
+	maxPushPreviewRunes = 120
+)
+
+// presenceKey marks "this user has a live realtime connection".
+func presenceKey(userID string) string { return "chat:online:" + userID }
+
+// MarkPresence records that a user's realtime channel became occupied or
+// vacated. Driven by Soketi's webhooks.
+func (s *ChatService) MarkPresence(ctx context.Context, userID string, online bool) {
+	if s.cache == nil {
+		return
+	}
+	if online {
+		s.cache.Client.Set(ctx, presenceKey(userID), "1", presenceTTL)
+		return
+	}
+	s.cache.Client.Del(ctx, presenceKey(userID))
+}
+
+// notifyNewMessage delivers a Web Push alert to participants who are not
+// currently connected.
+//
+// Deliberately NOT routed through NotificationService: chat must not create a
+// notifications row or land in the tray. It borrows only the preference matrix
+// so "Messages → Push" can be turned off like anything else.
+//
+// Four things suppress a push, in the order that costs least to check:
+// the thread is muted, the user opted out, they have a live connection, or one
+// already went out for this thread recently.
+func (s *ChatService) notifyNewMessage(ctx context.Context, conv *model.Conversation, senderID string, msg *model.Message) {
+	if s.push == nil || !s.push.Enabled() {
+		return
+	}
+	candidates, err := s.chat.PushCandidates(ctx, conv.ID, senderID)
+	if err != nil {
+		s.logger.Warn("chat: push candidates failed", zap.Error(err))
+		return
+	}
+
+	title, body := s.pushCopy(ctx, conv, senderID, msg)
+	now := time.Now()
+
+	for _, c := range candidates {
+		if c.MutedUntil != nil && c.MutedUntil.After(now) {
+			continue
+		}
+		if s.prefs != nil {
+			if off, err := s.prefs.DisabledSet(ctx, c.UserID); err == nil &&
+				prefOff(off, NotifChatMessage, repository.ChannelPush) {
+				continue
+			}
+		}
+		if s.online(ctx, c.UserID) {
+			continue
+		}
+		if !s.claimPushSlot(ctx, c.UserID, conv.ID) {
+			continue
+		}
+		s.push.SendToUser(ctx, c.UserID, PushPayload{
+			Title: title,
+			Body:  body,
+			URL:   "/chats/" + conv.ID,
+			// Same tag per thread: a later message replaces the earlier
+			// notification rather than stacking up.
+			Tag: "chat:" + conv.ID,
+		})
+	}
+}
+
+// pushCopy builds the lock-screen text.
+func (s *ChatService) pushCopy(ctx context.Context, conv *model.Conversation, senderID string, msg *model.Message) (title, body string) {
+	sender := "Someone"
+	if s.users != nil {
+		if u, err := s.users.FindByID(ctx, senderID); err == nil && u != nil {
+			if u.Name != "" {
+				sender = u.Name
+			} else if u.Handle != "" {
+				sender = "@" + u.Handle
+			}
+		}
+	}
+
+	title = sender
+	if conv.Kind == model.ConversationGroup && conv.Title != nil && *conv.Title != "" {
+		title = sender + " · " + *conv.Title
+	}
+
+	switch {
+	case strings.TrimSpace(msg.Body) != "":
+		body = truncateRunes(strings.TrimSpace(msg.Body), maxPushPreviewRunes)
+	case len(msg.Attachments) > 0:
+		body = "Sent an attachment"
+	default:
+		body = "Sent you a message"
+	}
+	return title, body
+}
+
+// online reports whether the user holds a live realtime connection.
+//
+// Best-effort and fails OPEN: if the cache is unavailable — or Soketi is not
+// configured to send channel_occupied — we treat them as offline and push. A
+// redundant notification while they are reading is a smaller failure than a
+// message that never arrives.
+func (s *ChatService) online(ctx context.Context, userID string) bool {
+	if s.cache == nil {
+		return false
+	}
+	n, err := s.cache.Client.Exists(ctx, presenceKey(userID)).Result()
+	return err == nil && n > 0
+}
+
+// claimPushSlot returns true for the first push to a thread within the window.
+// Fails OPEN for the same reason as online().
+func (s *ChatService) claimPushSlot(ctx context.Context, userID, conversationID string) bool {
+	if s.cache == nil {
+		return true
+	}
+	ok, err := s.cache.Client.SetNX(ctx,
+		"chat:pushed:"+userID+":"+conversationID, "1", pushDebounceWindow).Result()
+	if err != nil {
+		return true
+	}
+	return ok
+}
+
+func truncateRunes(s string, n int) string {
+	r := []rune(s)
+	if len(r) <= n {
+		return s
+	}
+	return strings.TrimSpace(string(r[:n])) + "…"
+}
+
 func (s *ChatService) publishReceipt(ctx context.Context, conversationID, userID, kind string, seq int64) {
 	if s.rt == nil {
 		return

internal/service/chat_service_test.go

@@ -7,10 +7,14 @@ import (
 	"errors"
 	"os"
 	"path/filepath"
+	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/joho/godotenv"
 	"github.com/prabhupadasays/spsays-bff/internal/model"
 	"github.com/prabhupadasays/spsays-bff/internal/repository"
@@ -39,7 +43,8 @@ func testChatService(t *testing.T) (*ChatService, *repository.ChatRepo, *reposit
 	// Nil realtime/R2/cache clients make publishing, uploads and caching
 	// no-ops — these tests are about the rules, and none of those services are
 	// reachable from a test run.
-	return NewChatService(chat, repository.NewBlockRepo(pg), nil, nil, nil, zap.NewNop()), chat, pg
+	return NewChatService(chat, repository.NewBlockRepo(pg), repository.NewPostRepo(pg),
+		nil, nil, nil, nil, nil, nil, zap.NewNop()), chat, pg
 }
 
 // chatUserWithHandle returns both identifiers: ids for authoring, handles for
@@ -1719,3 +1724,483 @@ func TestOpenDMByHandle(t *testing.T) {
 		t.Errorf("unknown handle: err = %v, want ErrUserGone", err)
 	}
 }
+
+// A quote renders as trusted in-app content, so its link must be unable to
+// leave the app. Anything that could navigate off-site has to be refused.
+func TestQuoteAttachment_RejectsOffSitePaths(t *testing.T) {
+	good := []string{
+		"/books/bg/2/13",
+		"/books/bg/2/13?hl=abc%20def",
+		"/u/someone/posts#post-1",
+		"/",
+	}
+	for _, p := range good {
+		q := QuoteAttachment{Path: p, Title: "Bg 2.13", Text: "the passage"}
+		if !q.valid() {
+			t.Errorf("valid() = false for in-app path %q", p)
+		}
+	}
+
+	bad := map[string]string{
+		"absolute http":      "http://evil.example/x",
+		"absolute https":     "https://evil.example/x",
+		"protocol relative":  "//evil.example/x",
+		"backslash relative": "/\\evil.example/x",
+		"scheme only":        "javascript:alert(1)",
+		"data uri":           "data:text/html,<script>",
+		"no leading slash":   "books/bg/2/13",
+		"empty":              "",
+		"newline injection":  "/books\n/evil",
+		"tab injection":      "/books\t/evil",
+	}
+	for name, p := range bad {
+		q := QuoteAttachment{Path: p, Title: "Bg 2.13", Text: "the passage"}
+		if q.valid() {
+			t.Errorf("%s: valid() = true for %q — must be refused", name, p)
+		}
+	}
+
+	// A quote with no passage is not a quote.
+	if (&QuoteAttachment{Path: "/books/bg/2/13", Text: "   "}).valid() {
+		t.Error("valid() = true for a blank passage")
+	}
+	// Oversized fields are refused rather than truncated.
+	long := strings.Repeat("x", maxQuoteTextRunes+1)
+	if (&QuoteAttachment{Path: "/books/bg/2/13", Text: long}).valid() {
+		t.Error("valid() = true for an oversized passage")
+	}
+}
+
+// makePost inserts a post owned by userID and returns its id.
+func makePost(t *testing.T, pg *repository.Postgres, userID, caption string) string {
+	t.Helper()
+	var id string
+	if err := pg.Pool.QueryRow(context.Background(), `
+		INSERT INTO posts (user_id, caption, source_url, source_title)
+		VALUES ($1, $2, '/books/bg/2/13', 'Bhagavad-gītā 2.13') RETURNING id`,
+		userID, caption).Scan(&id); err != nil {
+		t.Fatalf("create post: %v", err)
+	}
+	t.Cleanup(func() {
+		_, _ = pg.Pool.Exec(context.Background(), `DELETE FROM posts WHERE id = $1`, id)
+	})
+	return id
+}
+
+func entityAttachment(t *testing.T, msgs []model.Message) *model.Attachment {
+	t.Helper()
+	for i := range msgs {
+		for j := range msgs[i].Attachments {
+			if msgs[i].Attachments[j].Kind == model.AttachmentEntity {
+				return &msgs[i].Attachments[j]
+			}
+		}
+	}
+	return nil
+}
+
+// Sharing is a reference, not a copy: access is decided when the message is
+// READ, per person. A post that stops being visible must stop rendering, even
+// though the snapshot taken at send time still sits in the row.
+func TestSharedPost_ResolvedPerViewerAtReadTime(t *testing.T) {
+	svc, _, pg := testChatService(t)
+	ctx := context.Background()
+	a, b := chatUser(t, pg, "A"), chatUser(t, pg, "B")
+	conv, err := svc.OpenDM(ctx, a, b)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cleanupConv(t, pg, conv.ID)
+
+	postID := makePost(t, pg, a, "a passage worth sharing")
+	if _, err := svc.Send(ctx, a, conv.ID, SendRequest{
+		Entity: &EntityAttachment{Type: model.EntityPost, ID: postID},
+	}); err != nil {
+		t.Fatalf("share post: %v", err)
+	}
+
+	// Recipient sees the card.
+	msgs, err := svc.Messages(ctx, b, conv.ID, 0, 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	att := entityAttachment(t, msgs)
+	if att == nil {
+		t.Fatal("no entity attachment on the shared message")
+	}
+	if att.EntityUnavailable || len(att.Entity) == 0 {
+		t.Fatalf("recipient should see the card: unavailable=%v entity=%s",
+			att.EntityUnavailable, att.Entity)
+	}
+
+	// Author hides the post → it must stop resolving for the recipient...
+	if _, err := pg.Pool.Exec(ctx, `UPDATE posts SET is_hidden = true WHERE id = $1`, postID); err != nil {
+		t.Fatal(err)
+	}
+	msgs, err = svc.Messages(ctx, b, conv.ID, 0, 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	att = entityAttachment(t, msgs)
+	if att == nil || !att.EntityUnavailable {
+		t.Error("a hidden post must render as unavailable to the recipient")
+	}
+	if att != nil && len(att.Entity) != 0 {
+		t.Errorf("stale snapshot leaked after the post was hidden: %s", att.Entity)
+	}
+
+	// ...while its owner still sees it, matching the feed's rules.
+	msgs, err = svc.Messages(ctx, a, conv.ID, 0, 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	att = entityAttachment(t, msgs)
+	if att == nil || att.EntityUnavailable {
+		t.Error("the owner should still see their own hidden post")
+	}
+}
+
+// You cannot share what you were never allowed to read.
+func TestSharedPost_SenderMustSeeItFirst(t *testing.T) {
+	svc, _, pg := testChatService(t)
+	ctx := context.Background()
+	a, b, stranger := chatUser(t, pg, "A"), chatUser(t, pg, "B"), chatUser(t, pg, "C")
+	conv, err := svc.OpenDM(ctx, a, b)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cleanupConv(t, pg, conv.ID)
+
+	hidden := makePost(t, pg, stranger, "not yours to share")
+	if _, err := pg.Pool.Exec(ctx, `UPDATE posts SET is_hidden = true WHERE id = $1`, hidden); err != nil {
+		t.Fatal(err)
+	}
+	msg, err := svc.Send(ctx, a, conv.ID, SendRequest{
+		Body:   "look at this",
+		Entity: &EntityAttachment{Type: model.EntityPost, ID: hidden},
+	})
+	if err != nil {
+		t.Fatalf("send: %v", err)
+	}
+	full, err := svc.Messages(ctx, a, conv.ID, 0, 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if att := entityAttachment(t, full); att != nil {
+		t.Errorf("a post the sender cannot see must not be attached (msg %s)", msg.ID)
+	}
+}
+
+// queryCounter counts every statement the pool issues.
+type queryCounter struct{ n atomic.Int64 }
+
+func (q *queryCounter) TraceQueryStart(ctx context.Context, _ *pgx.Conn, _ pgx.TraceQueryStartData) context.Context {
+	q.n.Add(1)
+	return ctx
+}
+func (q *queryCounter) TraceQueryEnd(context.Context, *pgx.Conn, pgx.TraceQueryEndData) {}
+
+// Resolving shared posts must cost the same whether a page carries one share
+// or many — otherwise a thread full of shared posts turns one screen into
+// dozens of round-trips.
+func TestSharedPosts_ResolutionDoesNotScaleWithShareCount(t *testing.T) {
+	uri := os.Getenv("POSTGRES_URI")
+	if uri == "" {
+		_ = godotenv.Load(filepath.Join("..", "..", ".env"))
+		uri = os.Getenv("POSTGRES_URI")
+	}
+	if uri == "" {
+		t.Skip("POSTGRES_URI not set")
+	}
+	ctx := context.Background()
+
+	cfg, err := pgxpool.ParseConfig(uri)
+	if err != nil {
+		t.Fatal(err)
+	}
+	counter := &queryCounter{}
+	cfg.ConnConfig.Tracer = counter
+	pool, err := pgxpool.NewWithConfig(ctx, cfg)
+	if err != nil {
+		t.Skipf("postgres unreachable: %v", err)
+	}
+	t.Cleanup(pool.Close)
+
+	pg := &repository.Postgres{Pool: pool}
+	svc := NewChatService(repository.NewChatRepo(pg), repository.NewBlockRepo(pg),
+		repository.NewPostRepo(pg), nil, nil, nil, nil, nil, nil, zap.NewNop())
+
+	// measure returns the statement count for reading a thread carrying
+	// `shares` distinct shared posts.
+	measure := func(shares int) int {
+		a, b := chatUser(t, pg, "A"), chatUser(t, pg, "B")
+		conv, err := svc.OpenDM(ctx, a, b)
+		if err != nil {
+			t.Fatal(err)
+		}
+		cleanupConv(t, pg, conv.ID)
+		for i := 0; i < shares; i++ {
+			id := makePost(t, pg, a, "shared")
+			if _, err := svc.Send(ctx, a, conv.ID, SendRequest{
+				Entity: &EntityAttachment{Type: model.EntityPost, ID: id},
+			}); err != nil {
+				t.Fatal(err)
+			}
+		}
+		// Warm the pool so connection setup isn't counted.
+		if _, err := svc.Messages(ctx, b, conv.ID, 0, 50); err != nil {
+			t.Fatal(err)
+		}
+
+		before := counter.n.Load()
+		msgs, err := svc.Messages(ctx, b, conv.ID, 0, 50)
+		if err != nil {
+			t.Fatal(err)
+		}
+		after := counter.n.Load()
+
+		// Guard the measurement: the cards must actually have resolved, or a
+		// low count would just mean nothing was looked up.
+		resolved := 0
+		for i := range msgs {
+			for j := range msgs[i].Attachments {
+				if msgs[i].Attachments[j].Kind == model.AttachmentEntity &&
+					!msgs[i].Attachments[j].EntityUnavailable {
+					resolved++
+				}
+			}
+		}
+		if resolved != shares {
+			t.Fatalf("resolved %d of %d shared posts", resolved, shares)
+		}
+		return int(after - before)
+	}
+
+	one, many := measure(1), measure(8)
+	t.Logf("queries: 1 share = %d, 8 shares = %d", one, many)
+	if many != one {
+		t.Errorf("reading 8 shares cost %d queries vs %d for 1 — resolution is still per-share",
+			many, one)
+	}
+}
+
+// The nav badge counts threads waiting on you — and mute is meant to stop a
+// thread demanding attention, so a muted one must not light it.
+func TestUnreadCount_CountsThreadsAndSkipsMuted(t *testing.T) {
+	svc, _, pg := testChatService(t)
+	ctx := context.Background()
+	me, a, b := chatUser(t, pg, "Me"), chatUser(t, pg, "A"), chatUser(t, pg, "B")
+
+	if n, err := svc.UnreadCount(ctx, me); err != nil || n != 0 {
+		t.Fatalf("fresh account: count = %d, err = %v; want 0", n, err)
+	}
+
+	withA, err := svc.OpenDM(ctx, a, me)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cleanupConv(t, pg, withA.ID)
+	withB, err := svc.OpenDM(ctx, b, me)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cleanupConv(t, pg, withB.ID)
+
+	// Two messages in one thread still count as one waiting thread.
+	for _, body := range []string{"hare", "krsna"} {
+		if _, err := svc.Send(ctx, a, withA.ID, SendRequest{Body: body}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := svc.Send(ctx, b, withB.ID, SendRequest{Body: "hello"}); err != nil {
+		t.Fatal(err)
+	}
+	if n, _ := svc.UnreadCount(ctx, me); n != 2 {
+		t.Errorf("two threads with unread: count = %d, want 2", n)
+	}
+
+	// Muting one drops it from the badge...
+	if _, err := svc.SetMute(ctx, me, withA.ID, MuteAlways); err != nil {
+		t.Fatal(err)
+	}
+	if n, _ := svc.UnreadCount(ctx, me); n != 1 {
+		t.Errorf("after muting one: count = %d, want 1", n)
+	}
+	// ...without marking it read: the inbox still shows its unread.
+	inbox, err := svc.Inbox(ctx, me, "chats", 10)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var mutedUnread int64
+	for _, it := range inbox {
+		if it.ID == withA.ID {
+			mutedUnread = it.Unread
+		}
+	}
+	if mutedUnread != 2 {
+		t.Errorf("muted thread's inbox unread = %d, want 2 — mute silences, it does not read", mutedUnread)
+	}
+
+	// Reading the other clears the badge entirely.
+	if err := svc.MarkRead(ctx, me, withB.ID, 1); err != nil {
+		t.Fatal(err)
+	}
+	if n, _ := svc.UnreadCount(ctx, me); n != 0 {
+		t.Errorf("after reading: count = %d, want 0", n)
+	}
+}
+
+// A chat push must not fire when the thread is muted or the user opted out.
+// Both are the difference between a useful alert and an uninstall.
+func TestChatPush_SuppressedByMuteAndPreference(t *testing.T) {
+	svc, _, pg := testChatService(t)
+	ctx := context.Background()
+	prefs := repository.NewNotificationPrefsRepo(pg)
+
+	t.Run("muted thread does not push", func(t *testing.T) {
+		me, them := chatUser(t, pg, "Me"), chatUser(t, pg, "Them")
+		conv, err := svc.OpenDM(ctx, them, me)
+		if err != nil {
+			t.Fatal(err)
+		}
+		cleanupConv(t, pg, conv.ID)
+		if _, err := svc.SetMute(ctx, me, conv.ID, MuteAlways); err != nil {
+			t.Fatal(err)
+		}
+		cands, err := svc.chat.PushCandidates(ctx, conv.ID, them)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for _, c := range cands {
+			if c.UserID == me && (c.MutedUntil == nil || !c.MutedUntil.After(time.Now())) {
+				t.Error("muted participant should carry a future muted_until")
+			}
+		}
+	})
+
+	t.Run("push opt-out is recorded for the chat kind", func(t *testing.T) {
+		me := chatUser(t, pg, "Me")
+		if err := prefs.Disable(ctx, me,
+			[]string{NotifChatMessage}, []string{repository.ChannelPush}); err != nil {
+			t.Fatal(err)
+		}
+		off, err := prefs.DisabledSet(ctx, me)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !prefOff(off, NotifChatMessage, repository.ChannelPush) {
+			t.Error("disabling Messages→Push should be visible to the chat push path")
+		}
+		// Turning off chat push must not silence anything else.
+		if prefOff(off, NotifPostLike, repository.ChannelPush) {
+			t.Error("disabling chat push also disabled post likes")
+		}
+	})
+
+	t.Run("sender is never a push candidate", func(t *testing.T) {
+		me, them := chatUser(t, pg, "Me"), chatUser(t, pg, "Them")
+		conv, err := svc.OpenDM(ctx, me, them)
+		if err != nil {
+			t.Fatal(err)
+		}
+		cleanupConv(t, pg, conv.ID)
+		cands, err := svc.chat.PushCandidates(ctx, conv.ID, me)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for _, c := range cands {
+			if c.UserID == me {
+				t.Error("the sender must not be pushed their own message")
+			}
+		}
+		if len(cands) != 1 {
+			t.Errorf("candidates = %d, want 1 (the recipient)", len(cands))
+		}
+	})
+
+	t.Run("a departed member stops receiving push", func(t *testing.T) {
+		owner, a, b := chatUser(t, pg, "O"), chatUser(t, pg, "A"), chatUser(t, pg, "B")
+		conv, err := svc.CreateGroup(ctx, owner, "Kīrtana",
+			[]string{handleOf(t, pg, a), handleOf(t, pg, b)})
+		if err != nil {
+			t.Fatal(err)
+		}
+		cleanupConv(t, pg, conv.ID)
+		if err := svc.RemoveMember(ctx, b, conv.ID, handleOf(t, pg, b)); err != nil {
+			t.Fatal(err)
+		}
+		cands, err := svc.chat.PushCandidates(ctx, conv.ID, owner)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for _, c := range cands {
+			if c.UserID == b {
+				t.Error("someone who left the group must not be pushed")
+			}
+		}
+	})
+}
+
+// The three tick states have to be reachable in order: one tick on send, two
+// once it reaches the other device, blue once they read it. The middle state
+// only exists if delivery is acked separately from reading.
+func TestReceipts_TickProgression(t *testing.T) {
+	svc, _, pg := testChatService(t)
+	ctx := context.Background()
+	a, b := chatUser(t, pg, "A"), chatUser(t, pg, "B")
+	conv, err := svc.OpenDM(ctx, a, b)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cleanupConv(t, pg, conv.ID)
+
+	msg, err := svc.Send(ctx, a, conv.ID, SendRequest{Body: "hare kṛṣṇa"})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Sent: nothing acked yet — one tick.
+	rc, err := svc.Receipts(ctx, a, conv.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if rc.DeliveredSeq >= msg.Seq || rc.ReadSeq >= msg.Seq {
+		t.Fatalf("just sent: %+v — expected neither delivered nor read", rc)
+	}
+
+	// Delivered: reached their device, still unread — two grey ticks.
+	if err := svc.MarkDelivered(ctx, b, conv.ID, msg.Seq); err != nil {
+		t.Fatal(err)
+	}
+	rc, _ = svc.Receipts(ctx, a, conv.ID)
+	if rc.DeliveredSeq < msg.Seq {
+		t.Errorf("after delivery ack: delivered_seq = %d, want >= %d", rc.DeliveredSeq, msg.Seq)
+	}
+	if rc.ReadSeq >= msg.Seq {
+		t.Errorf("delivery must not imply read: %+v", rc)
+	}
+
+	// Read: blue.
+	if err := svc.MarkRead(ctx, b, conv.ID, msg.Seq); err != nil {
+		t.Fatal(err)
+	}
+	rc, _ = svc.Receipts(ctx, a, conv.ID)
+	if rc.ReadSeq < msg.Seq {
+		t.Errorf("after read: read_seq = %d, want >= %d", rc.ReadSeq, msg.Seq)
+	}
+
+	// Reading implies delivery, so someone who was offline and opens the thread
+	// can't end up read-but-undelivered.
+	if rc.DeliveredSeq < rc.ReadSeq {
+		t.Errorf("delivered_seq (%d) fell behind read_seq (%d)", rc.DeliveredSeq, rc.ReadSeq)
+	}
+
+	// Acks must never move a cursor backwards.
+	if err := svc.MarkDelivered(ctx, b, conv.ID, 1); err != nil {
+		t.Fatal(err)
+	}
+	back, _ := svc.Receipts(ctx, a, conv.ID)
+	if back.DeliveredSeq < rc.DeliveredSeq || back.ReadSeq < rc.ReadSeq {
+		t.Errorf("a stale ack rewound the cursors: %+v then %+v", rc, back)
+	}
+}

internal/service/notification_service.go

@@ -44,6 +44,11 @@ const (
 	// disciples. The actor is the disciple, so it renders and links exactly
 	// like the other actor-driven kinds.
 	NotifDiscipleJoined = "disciple_joined"
+	// NotifChatMessage exists ONLY as a preference key. Chat never creates a
+	// notifications row and never appears in the tray — a new message
+	// highlights its inbox row instead. The kind is here so the settings
+	// matrix can offer a push opt-out alongside everything else.
+	NotifChatMessage = "chat_message"
 
 	// Reminder kinds are delivered live (email + push + a websocket toast)
 	// but NOT persisted — they're ephemeral nudges, not history.
@@ -60,6 +65,9 @@ var PrefCategories = map[string][]string{
 	"post_interactions": {NotifPostLike, NotifPostMention, NotifPostComment, NotifCommentReply},
 	"sadhna_reminders":  {NotifChantReminder, NotifQuoteReminder, NotifChantIncomplete},
 	"community":         {NotifDiscipleJoined},
+	// Push only: in-app delivery for chat is the inbox itself, and there is
+	// no such thing as an emailed chat message.
+	"messages": {NotifChatMessage},
 }
 
 // PrefChannels is every delivery channel a matrix cell can address.

migrations/0051_quote_attachment.down.sql

@@ -0,0 +1,8 @@
+-- Rows of the new kind must go before the old constraint can be restored.
+DELETE FROM message_attachments WHERE kind = 'quote';
+
+ALTER TABLE message_attachments DROP CONSTRAINT IF EXISTS message_attachments_kind_check;
+
+ALTER TABLE message_attachments
+    ADD CONSTRAINT message_attachments_kind_check
+    CHECK (kind IN ('image', 'video', 'audio', 'file', 'link', 'entity'));

migrations/0051_quote_attachment.up.sql

@@ -0,0 +1,18 @@
+-- A shared passage from the library, sent into a chat.
+--
+-- Modelled as an attachment rather than message text so it can be rendered as
+-- a quote and navigated back to. It reuses the columns the link card already
+-- has, since the shape is the same thing said differently:
+--
+--   url         → the ?hl= deep link back to the exact phrase
+--   title       → the source reference ("Bhagavad-gītā 2.13")
+--   description → the quoted passage itself
+--
+-- It is a distinct kind rather than a 'link' because it renders as a quote,
+-- not a preview card, and because its URL is always our own — the service
+-- refuses any other origin.
+ALTER TABLE message_attachments DROP CONSTRAINT IF EXISTS message_attachments_kind_check;
+
+ALTER TABLE message_attachments
+    ADD CONSTRAINT message_attachments_kind_check
+    CHECK (kind IN ('image', 'video', 'audio', 'file', 'link', 'entity', 'quote'));

migrations/0052_broadcasts.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS broadcast_views;
+DROP TABLE IF EXISTS broadcasts;

migrations/0052_broadcasts.up.sql

@@ -0,0 +1,44 @@
+-- Admin broadcasts — product updates and announcements, shown once per user.
+--
+-- On "no history": these deliberately create no notification rows and never
+-- appear in the tray. But "exactly once per user" cannot be done without
+-- storing something — the app has to know whether YOU have seen it. So two
+-- tables: the announcement, and who has dismissed it. Nothing per-user is
+-- written until that person actually sees it.
+CREATE TABLE IF NOT EXISTS broadcasts (
+    id          UUID         PRIMARY KEY DEFAULT gen_random_uuid(),
+    title       VARCHAR(200) NOT NULL,
+    body        TEXT         NOT NULL DEFAULT '',
+    -- Formatting as offset/length spans, exactly like chat messages: the same
+    -- parser produces them and the same component renders them, so no HTML is
+    -- stored and nothing needs sanitising at render time.
+    entities    JSONB        NOT NULL DEFAULT '[]'::jsonb,
+    -- R2 key for the banner. Public bucket: an announcement is not private,
+    -- and a signed URL would expire while the modal sat open.
+    banner_key  TEXT,
+    cta_label   VARCHAR(60),
+    -- App-relative path or absolute http(s) URL; validated in the service.
+    cta_url     TEXT,
+    created_by  UUID         REFERENCES users(id) ON DELETE SET NULL,
+    -- NULL = draft. Nothing is delivered until this is set.
+    published_at TIMESTAMPTZ,
+    -- After this, the announcement stops being offered. This is what keeps a
+    -- months-old update from greeting someone who signs up tomorrow.
+    expires_at   TIMESTAMPTZ,
+    created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at   TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+-- "What is live right now?" — the app's pending lookup and the admin list.
+CREATE INDEX IF NOT EXISTS broadcasts_live_idx
+    ON broadcasts (published_at DESC) WHERE published_at IS NOT NULL;
+
+-- One row per person who has seen an announcement. Written on dismissal, which
+-- is what makes "at most once" true across devices — localStorage would show
+-- it again on every new browser.
+CREATE TABLE IF NOT EXISTS broadcast_views (
+    broadcast_id UUID        NOT NULL REFERENCES broadcasts(id) ON DELETE CASCADE,
+    user_id      UUID        NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    seen_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (broadcast_id, user_id)
+);