feat: enforce current workspace membership on owner-addressed realtime events and notification ingress, stored feeds, deferred email and push delivery

cmd/backend/main.go

@@ -527,11 +527,11 @@ func main() {
 				errs.CaptureFatal(err)
 				log.Fatal("Failed to provision Pub/Sub topics/subscriptions: ", err)
 			}
-			streamingPublisher = pubsub.NewStreamingPublisher(pubsubClient)
+			streamingPublisher = pubsub.NewStreamingPublisher(pubsubClient, repository.NewEventAudienceRepository(primaryDB.Pool))
 			log.Println("Realtime events published to Google Pub/Sub")
 		}
 		if streamingPublisher == nil {
-			streamingPublisher = pubsub.NewStreamingPublisher(pubsub.NewRedisBus(cache.Client, ""))
+			streamingPublisher = pubsub.NewStreamingPublisher(pubsub.NewRedisBus(cache.Client, ""), repository.NewEventAudienceRepository(primaryDB.Pool))
 			log.Println("Realtime events bridged over Redis (Pub/Sub disabled)")
 		}
 

cmd/consumer/main.go

@@ -194,10 +194,10 @@ func main() {
 			errs.CaptureFatal(err)
 			log.Fatal("Failed to provision Pub/Sub topics/subscriptions: ", err)
 		}
-		streamingPublisher = pubsub.NewStreamingPublisher(pubsubClient)
+		streamingPublisher = pubsub.NewStreamingPublisher(pubsubClient, repository.NewEventAudienceRepository(primaryDB.Pool))
 	}
 	if streamingPublisher == nil {
-		streamingPublisher = pubsub.NewStreamingPublisher(pubsub.NewRedisBus(redisCache.Client, ""))
+		streamingPublisher = pubsub.NewStreamingPublisher(pubsub.NewRedisBus(redisCache.Client, ""), repository.NewEventAudienceRepository(primaryDB.Pool))
 		log.Println("Realtime events bridged over Redis (Pub/Sub disabled)")
 	}
 

internal/app/consumer/sync_truth_test.go

@@ -3,6 +3,7 @@ package jobs
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"log"
@@ -95,12 +96,30 @@ func (r *removeSyncWarmup) GetWarmupReceived(_ context.Context, mailbox, id uuid
 }
 
 func (b *removeSyncBus) Publish(_ context.Context, _ string, data any, _ map[string]string) error {
-	if event, ok := data.(*pubsub.EmailInboxEvent); ok && event.EventType == pubsub.EventEmailDeleted {
+	wire, err := json.Marshal(data)
+	if err != nil {
+		return err
+	}
+	var event pubsub.EmailInboxEvent
+	if err := json.Unmarshal(wire, &event); err != nil {
+		return err
+	}
+	if event.EventType == pubsub.EventEmailDeleted {
 		b.deleted++
 	}
 	return nil
 }
 
+type removeSyncAudience struct{ org uuid.UUID }
+
+func (a removeSyncAudience) ResourceOrganization(context.Context, string, uuid.UUID) (uuid.UUID, error) {
+	return a.org, nil
+}
+
+func (a removeSyncAudience) CanAddressUser(context.Context, uuid.UUID, uuid.UUID) (bool, error) {
+	return true, nil
+}
+
 func TestSyncTruthRemovalWaitsForScopedPendingArrivals(t *testing.T) {
 	user, mailbox, id := uuid.New(), uuid.New(), uuid.New()
 	lookupErr := errors.New("pending marker unavailable")
@@ -119,7 +138,7 @@ func TestSyncTruthRemovalWaitsForScopedPendingArrivals(t *testing.T) {
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			inbox, bus := &removeSyncInbox{}, &removeSyncBus{}
-			s := &JobsService{UniboxRepository: inbox, ArrivalOutbox: &syncTruthOutbox{user: user, mailbox: mailbox, id: id, pending: tc.pending, err: tc.err}, EmailRepository: newEmailAccountRepo{}, StreamingPublisher: pubsub.NewStreamingPublisher(bus)}
+			s := &JobsService{UniboxRepository: inbox, ArrivalOutbox: &syncTruthOutbox{user: user, mailbox: mailbox, id: id, pending: tc.pending, err: tc.err}, EmailRepository: newEmailAccountRepo{}, StreamingPublisher: pubsub.NewStreamingPublisher(bus, removeSyncAudience{org: uuid.New()})}
 			err := s.HandleRemoveEmail(t.Context(), &models.JobEventRemoveEmail{UserID: tc.owner, EmailID: tc.account, ID: tc.message})
 			if !errors.Is(err, tc.want) {
 				t.Fatalf("removal error=%v want=%v", err, tc.want)
@@ -139,7 +158,7 @@ func TestSyncTruthRemovalDeleteFailureRetriesBeforeNotification(t *testing.T) {
 	failure := errors.New("unibox delete unavailable")
 	inbox, bus := &removeSyncInbox{err: failure}, &removeSyncBus{}
 	warmup := &removeSyncWarmup{}
-	s := &JobsService{UniboxRepository: inbox, WarmupRepo: warmup, EmailRepository: newEmailAccountRepo{}, StreamingPublisher: pubsub.NewStreamingPublisher(bus)}
+	s := &JobsService{UniboxRepository: inbox, WarmupRepo: warmup, EmailRepository: newEmailAccountRepo{}, StreamingPublisher: pubsub.NewStreamingPublisher(bus, removeSyncAudience{org: uuid.New()})}
 	e := &models.JobEventRemoveEmail{UserID: uuid.New(), EmailID: uuid.New(), ID: uuid.New()}
 	if err := s.HandleRemoveEmail(t.Context(), e); !errors.Is(err, failure) {
 		t.Fatalf("delete failure acknowledged: %v", err)

internal/app/notification/audience_test.go

@@ -0,0 +1,107 @@
+package notification
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+type audienceMembers struct {
+	members []models.OrganizationMember
+	err     error
+}
+
+func (m *audienceMembers) GetMembers(context.Context, uuid.UUID) ([]models.OrganizationMember, error) {
+	return m.members, m.err
+}
+
+func acceptedMember(userID uuid.UUID) *audienceMembers {
+	now := time.Now()
+	return &audienceMembers{members: []models.OrganizationMember{{UserID: userID, AcceptedAt: &now, AccessScope: models.AccessScopeWorkspace}}}
+}
+
+func TestWorkspaceNotificationIngress(t *testing.T) {
+	userID, orgID := uuid.New(), uuid.New()
+	for _, state := range []string{"unrestricted", "restricted", "removed", "invited", "error", "missing", "personal", "unscoped workspace"} {
+		t.Run(state, func(t *testing.T) {
+			members := acceptedMember(userID)
+			var lookup MemberLookup = members
+			org, category := &orgID, models.NotifCampaignPaused
+			want := false
+			switch state {
+			case "unrestricted":
+				want = true
+			case "restricted":
+				members.members[0].AccessScope = models.AccessScopeRestricted
+			case "removed":
+				members.members = nil
+			case "invited":
+				members.members[0].AcceptedAt = nil
+			case "error":
+				members.err = errors.New("resolver unavailable")
+			case "missing":
+				lookup = nil
+			case "personal":
+				org, category, lookup, want = nil, models.NotifSecuritySignIn, nil, true
+			case "unscoped workspace":
+				org = nil
+			}
+			repo := &messageNotificationRepo{channels: models.ChannelPrefs{InApp: true, Email: true}}
+			s := &service{repo: repo, members: lookup}
+			created, err := s.notifyOneWithError(context.Background(), userID, org, nil, category, "private", "private", "", nil, "", false)
+			if (repo.created > 0) != want {
+				t.Fatalf("created=%v stored=%d want=%v err=%v", created, repo.created, want, err)
+			}
+			if (state == "error" || state == "missing") && err == nil {
+				t.Fatal("resolver failure must remain retryable")
+			}
+		})
+	}
+}
+
+func TestDeferredNotificationRevocation(t *testing.T) {
+	ctx := context.Background()
+	userID, orgID := uuid.New(), uuid.New()
+	members := acceptedMember(userID)
+	repo := &messageNotificationRepo{}
+	s := &service{repo: repo, members: members}
+	workspace := models.Notification{ID: uuid.New(), UserID: userID, OrganizationID: &orgID, Category: models.NotifCampaignPaused}
+	personal := models.Notification{ID: uuid.New(), UserID: userID, Category: models.NotifSecuritySignIn}
+	push := pendingPush{OrganizationID: &orgID, Title: "private"}
+	if !s.canDeliverPush(ctx, userID, workspace.Category, push) {
+		t.Fatal("unrestricted push must remain eligible")
+	}
+	for _, state := range []string{"restricted", "removed", "error", "missing"} {
+		t.Run(state, func(t *testing.T) {
+			members = acceptedMember(userID)
+			s.members = members
+			switch state {
+			case "restricted":
+				members.members[0].AccessScope = models.AccessScopeRestricted
+			case "removed":
+				members.members = nil
+			case "error":
+				members.err = errors.New("unavailable")
+			case "missing":
+				s.members = nil
+			}
+			kept := s.keepEligibleEmailMessages(ctx, []models.Notification{workspace, personal})
+			if len(kept) != 1 || kept[0].ID != personal.ID {
+				t.Fatalf("workspace email survived revocation: %+v", kept)
+			}
+			if s.canDeliverPush(ctx, userID, workspace.Category, push) {
+				t.Fatal("workspace push survived revocation")
+			}
+			if !s.canDeliverPush(ctx, userID, personal.Category, pendingPush{}) {
+				t.Fatal("personal push requires no workspace")
+			}
+		})
+	}
+	if s.canDeliverPush(ctx, userID, workspace.Category, pendingPush{Title: "legacy batch"}) {
+		t.Fatal("legacy queued workspace push without provenance must fail closed")
+	}
+}

internal/app/notification/email.go

@@ -134,28 +134,12 @@ func (s *service) sendGroupEmail(ctx context.Context, rows []models.Notification
 	if len(rows) == 0 {
 		return
 	}
-	var stillMember map[uuid.UUID]bool
-	if s.members != nil && rows[0].OrganizationID != nil {
-		if members, err := s.members.GetMembers(ctx, *rows[0].OrganizationID); err == nil {
-			stillMember = map[uuid.UUID]bool{}
-			for _, m := range members {
-				if m.AcceptedAt != nil {
-					stillMember[m.UserID] = true
-				}
-			}
-		}
-	}
-
 	to := make([]string, 0, len(rows))
 	seen := map[string]bool{}
 	kept := make([]models.Notification, 0, len(rows))
 	dropped := make([]uuid.UUID, 0)
 	budgetOK := map[uuid.UUID]bool{}
 	for _, n := range rows {
-		if stillMember != nil && !stillMember[n.UserID] {
-			dropped = append(dropped, n.ID)
-			continue
-		}
 		if _, checked := budgetOK[n.UserID]; !checked {
 			budgetOK[n.UserID] = !s.overEmailBudget(ctx, n.UserID)
 		}
@@ -262,7 +246,10 @@ func (s *service) keepEligibleEmailMessages(ctx context.Context, rows []models.N
 	kept := rows[:0:0]
 	var skipped, retry []uuid.UUID
 	for _, n := range rows {
-		allowed, err := s.messageEligibility(ctx, n.Category, n.UniboxEmailID)
+		allowed, err := s.recipientEligibility(ctx, n.UserID, n.OrganizationID, n.Category)
+		if err == nil && allowed {
+			allowed, err = s.messageEligibility(ctx, n.Category, n.UniboxEmailID)
+		}
 		if err != nil {
 			retry = append(retry, n.ID)
 		} else if allowed {

internal/app/notification/message_test.go

@@ -29,6 +29,8 @@ func (r *messageNotificationRepo) CanNotifyAboutMessage(context.Context, uuid.UU
 func (r *messageNotificationRepo) GetPreferences(context.Context, uuid.UUID) (*models.NotificationPreferences, error) {
 	prefs := models.DefaultNotificationPreferences()
 	prefs.InboundReply = models.CategoryPref{Enabled: true, Channels: r.channels}
+	prefs.CampaignPaused = models.CategoryPref{Enabled: true, Channels: r.channels}
+	prefs.SecuritySignIn = models.CategoryPref{Enabled: true, Channels: r.channels}
 	return &prefs, nil
 }
 
@@ -49,11 +51,12 @@ func (r *messageNotificationRepo) RequeueEmails(_ context.Context, ids []uuid.UU
 
 func TestEmailRechecksTheMessageAfterClaiming(t *testing.T) {
 	id, notificationID, actionID := uuid.New(), uuid.New(), uuid.New()
+	userID, orgID := uuid.New(), uuid.New()
 	r := &messageNotificationRepo{}
-	s := &service{repo: r}
+	s := &service{repo: r, members: acceptedMember(userID)}
 	kept := s.keepEligibleEmailMessages(context.Background(), []models.Notification{
-		{ID: notificationID, Category: models.NotifInboundReply, UniboxEmailID: &id},
-		{ID: actionID, Category: models.NotifInboxActionRequired, UniboxEmailID: &id},
+		{ID: notificationID, UserID: userID, OrganizationID: &orgID, Category: models.NotifInboundReply, UniboxEmailID: &id},
+		{ID: actionID, UserID: userID, OrganizationID: &orgID, Category: models.NotifInboxActionRequired, UniboxEmailID: &id},
 	})
 	if len(kept) != 1 || kept[0].ID != actionID || len(r.skipped) != 1 || r.skipped[0] != notificationID {
 		t.Fatalf("kept=%v, skipped=%v", kept, r.skipped)
@@ -61,7 +64,7 @@ func TestEmailRechecksTheMessageAfterClaiming(t *testing.T) {
 	r.skipped = nil
 	r.err = errors.New("database unavailable")
 	kept = s.keepEligibleEmailMessages(context.Background(), []models.Notification{
-		{ID: notificationID, Category: models.NotifInboundReply, UniboxEmailID: &id},
+		{ID: notificationID, UserID: userID, OrganizationID: &orgID, Category: models.NotifInboundReply, UniboxEmailID: &id},
 	})
 	if len(kept) != 0 || len(r.skipped) != 0 || len(r.retry) != 1 || r.retry[0] != notificationID {
 		t.Fatalf("eligibility lookup failure did not retry: kept=%v skipped=%v retry=%v", kept, r.skipped, r.retry)
@@ -83,9 +86,10 @@ func TestNotifyAboutMessageGatesEveryChannel(t *testing.T) {
 		{"human reply", true, nil, models.ChannelPrefs{InApp: true}},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
+			userID, orgID := uuid.New(), uuid.New()
 			r := &messageNotificationRepo{allowed: tc.allowed, err: tc.err, channels: tc.channels}
-			s := &service{repo: r}
-			s.NotifyAboutMessage(context.Background(), uuid.New(), nil, uuid.New(), models.NotifInboundReply, "Reply", "", "", nil)
+			s := &service{repo: r, members: acceptedMember(userID)}
+			s.NotifyAboutMessage(context.Background(), userID, &orgID, uuid.New(), models.NotifInboundReply, "Reply", "", "", nil)
 			if r.checks != 1 {
 				t.Fatalf("eligibility checks = %d, want 1", r.checks)
 			}

internal/app/notification/push.go

@@ -36,10 +36,11 @@ const (
 )
 
 type pendingPush struct {
-	Title     string     `json:"title"`
-	Body      string     `json:"body,omitempty"`
-	Link      string     `json:"link,omitempty"`
-	MessageID *uuid.UUID `json:"unibox_email_id,omitempty"`
+	OrganizationID *uuid.UUID `json:"organization_id,omitempty"`
+	Title          string     `json:"title"`
+	Body           string     `json:"body,omitempty"`
+	Link           string     `json:"link,omitempty"`
+	MessageID      *uuid.UUID `json:"unibox_email_id,omitempty"`
 }
 
 func pushWindow() time.Duration {
@@ -94,7 +95,7 @@ func dueKey() string                  { return pushKeyPrefix + ":due" }
 func (s *service) deliverPush(userID uuid.UUID, category models.NotificationCategory, p pendingPush) {
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
-	if !s.canPushMessage(ctx, category, p) {
+	if !s.canDeliverPush(ctx, userID, category, p) {
 		return
 	}
 
@@ -180,7 +181,7 @@ func (s *service) sendDigest(ctx context.Context, member string) {
 	items := make([]pendingPush, 0, len(raw))
 	for _, r := range raw {
 		var p pendingPush
-		if json.Unmarshal([]byte(r), &p) == nil && s.canPushMessage(ctx, category, p) {
+		if json.Unmarshal([]byte(r), &p) == nil && s.canDeliverPush(ctx, userID, category, p) {
 			items = append(items, p)
 		}
 	}
@@ -197,6 +198,11 @@ func (s *service) sendDigest(ctx context.Context, member string) {
 	s.sendPush(ctx, userID, category, n)
 }
 
+func (s *service) canDeliverPush(ctx context.Context, userID uuid.UUID, category models.NotificationCategory, p pendingPush) bool {
+	allowed, err := s.recipientEligibility(ctx, userID, p.OrganizationID, category)
+	return err == nil && allowed && s.canPushMessage(ctx, category, p)
+}
+
 func (s *service) canPushMessage(ctx context.Context, category models.NotificationCategory, p pendingPush) bool {
 	if (category == models.NotifInboundReply || category == models.NotifInboundOOO) &&
 		replyclassify.IsSystemReport(replyclassify.Input{Subject: p.Body}) {
@@ -258,7 +264,7 @@ func (s *service) sendPush(ctx context.Context, userID uuid.UUID, category model
 	if err != nil || len(tokens) == 0 {
 		return
 	}
-	if unread, cerr := s.repo.CountUnread(ctx, userID); cerr == nil {
+	if unread, cerr := s.UnreadCount(ctx, userID); cerr == nil {
 		n.Badge = &unread
 	}
 	for _, t := range tokens {

internal/app/notification/service.go

@@ -242,6 +242,10 @@ func (s *service) notifyOneWithError(ctx context.Context, userID uuid.UUID, orgI
 	if userID == uuid.Nil {
 		return false, nil
 	}
+	allowed, err := s.recipientEligibility(ctx, userID, orgID, category)
+	if err != nil || !allowed {
+		return false, err
+	}
 	prefs, err := s.repo.GetPreferences(ctx, userID)
 	if err != nil || prefs == nil {
 		return false, err
@@ -286,7 +290,7 @@ func (s *service) notifyOneWithError(ctx context.Context, userID uuid.UUID, orgI
 			return false, nil
 		}
 		if cerr == nil && created != nil && cat.Channels.InApp && s.publisher != nil {
-			s.publisher.PublishNotificationCreated(ctx, userID.String(), created.ID.String(), string(category), title, link)
+			s.publisher.PublishNotificationCreated(ctx, userID.String(), created.ID.String(), string(category), title, link, orgID)
 		}
 	}
 
@@ -300,7 +304,8 @@ func (s *service) notifyOneWithError(ctx context.Context, userID uuid.UUID, orgI
 		go func(parent context.Context) {
 			ctx, cancel := context.WithTimeout(context.WithoutCancel(parent), 8*time.Second)
 			defer cancel()
-			if !s.canNotifyMessage(ctx, category, uniboxEmailID) {
+			allowed, err := s.recipientEligibility(ctx, userID, &org, category)
+			if err != nil || !allowed || !s.canNotifyMessage(ctx, category, uniboxEmailID) {
 				return
 			}
 			if postOrg {
@@ -316,11 +321,33 @@ func (s *service) notifyOneWithError(ctx context.Context, userID uuid.UUID, orgI
 
 	// Push: immediate on a quiet window, digest-batched inside one (detached).
 	if cat.Channels.Push && s.push != nil && s.deviceTokens != nil && s.pushRedis != nil {
-		go s.deliverPush(userID, category, pendingPush{Title: title, Body: body, Link: link, MessageID: uniboxEmailID})
+		go s.deliverPush(userID, category, pendingPush{Title: title, Body: body, Link: link, MessageID: uniboxEmailID, OrganizationID: orgID})
 	}
 	return slackFired, nil
 }
 
+func (s *service) recipientEligibility(ctx context.Context, userID uuid.UUID, orgID *uuid.UUID, category models.NotificationCategory) (bool, error) {
+	if orgID == nil {
+		return category == models.NotifSecuritySignIn, nil
+	}
+	if *orgID == uuid.Nil {
+		return false, nil
+	}
+	if s.members == nil {
+		return false, errors.New("notification: member resolver unavailable")
+	}
+	members, err := s.members.GetMembers(ctx, *orgID)
+	if err != nil {
+		return false, err
+	}
+	for _, m := range members {
+		if m.UserID == userID {
+			return m.AcceptedAt != nil && !m.IsRestricted(), nil
+		}
+	}
+	return false, nil
+}
+
 func (s *service) canNotifyMessage(ctx context.Context, category models.NotificationCategory, messageID *uuid.UUID) bool {
 	allowed, err := s.messageEligibility(ctx, category, messageID)
 	return err == nil && allowed

internal/infrastructure/pubsub/audience.go

@@ -0,0 +1,114 @@
+package pubsub
+
+import (
+	"context"
+	"encoding/json"
+
+	"github.com/google/uuid"
+)
+
+type AudienceResolver interface {
+	ResourceOrganization(ctx context.Context, kind string, id uuid.UUID) (uuid.UUID, error)
+	CanAddressUser(ctx context.Context, orgID, userID uuid.UUID) (bool, error)
+}
+
+// publish keeps workspace payloads on authorized org/resource routes when the owner shortcut is unavailable.
+func (p *StreamingPublisher) publish(ctx context.Context, topic string, event any, attrs map[string]string) error {
+	data, err := json.Marshal(event)
+	if err != nil {
+		return err
+	}
+	var fields map[string]json.RawMessage
+	if err := json.Unmarshal(data, &fields); err != nil {
+		return err
+	}
+	if fields == nil {
+		return nil
+	}
+	value := func(key string) string {
+		var v string
+		_ = json.Unmarshal(fields[key], &v)
+		return v
+	}
+	user := value("user_id")
+	if user == "" {
+		user = attrs["user_id"]
+	}
+	if user == "" || user == uuid.Nil.String() {
+		return p.client.Publish(ctx, topic, event, attrs)
+	}
+	org := value("org_id")
+	if org == "" {
+		org = value("organization_id")
+	}
+	if org == "" {
+		org = attrs["org_id"]
+	}
+	personal := org == "" && (value("event_type") == string(EventSessionsRevoked) ||
+		(value("event_type") == string(EventNotificationCreated) && value("category") == "security_new_signin"))
+	if personal {
+		return p.client.Publish(ctx, topic, event, attrs)
+	}
+	if p.audience == nil {
+		return nil
+	}
+	orgID, parseErr := uuid.Parse(org)
+	resolved := parseErr == nil && orgID != uuid.Nil
+	seen := make(map[string]bool)
+	for _, resource := range []struct{ key, kind string }{
+		{"campaign_id", "campaign"}, {"email_account_id", "mailbox"}, {"account_id", "mailbox"},
+		{"email_id", "mailbox"}, {"task_id", "task"}, {"booking_id", "booking"}, {"contact_id", "contact"},
+	} {
+		id := value(resource.key)
+		if id == "" {
+			id = attrs[resource.key]
+		}
+		if id == "" || id == uuid.Nil.String() {
+			continue
+		}
+		resourceID, err := uuid.Parse(id)
+		if err != nil {
+			return nil
+		}
+		key := resource.kind + ":" + resourceID.String()
+		if seen[key] {
+			continue
+		}
+		seen[key] = true
+		resourceOrg, err := p.audience.ResourceOrganization(ctx, resource.kind, resourceID)
+		if err != nil || resourceOrg == uuid.Nil {
+			return nil
+		}
+		if resolved && orgID != resourceOrg {
+			return nil
+		}
+		orgID, resolved = resourceOrg, true
+	}
+	userID, err := uuid.Parse(user)
+	allowed := false
+	if err == nil && resolved {
+		allowed, err = p.audience.CanAddressUser(ctx, orgID, userID)
+		if err != nil {
+			return nil
+		}
+	}
+	if !resolved {
+		return nil
+	}
+	if value("event_type") == string(EventNotificationCreated) {
+		if !allowed {
+			return nil
+		}
+		delete(fields, "org_id")
+		delete(fields, "organization_id")
+		delete(attrs, "org_id")
+		return p.client.Publish(ctx, topic, fields, attrs)
+	}
+	if !allowed {
+		delete(fields, "user_id")
+		delete(attrs, "user_id")
+	}
+	fields["org_id"], _ = json.Marshal(orgID.String())
+	attrs["org_id"] = orgID.String()
+	return p.client.Publish(ctx, topic, fields, attrs)
+}

internal/infrastructure/pubsub/audience_test.go

@@ -0,0 +1,130 @@
+package pubsub
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+
+	"github.com/google/uuid"
+)
+
+type audienceBus struct {
+	events []map[string]any
+	attrs  []map[string]string
+}
+
+func (b *audienceBus) Publish(_ context.Context, _ string, event interface{}, attrs map[string]string) error {
+	data, err := json.Marshal(event)
+	if err != nil {
+		return err
+	}
+	var fields map[string]any
+	if err := json.Unmarshal(data, &fields); err != nil {
+		return err
+	}
+	b.events = append(b.events, fields)
+	b.attrs = append(b.attrs, attrs)
+	return nil
+}
+
+type audienceStub struct {
+	org     uuid.UUID
+	allowed bool
+	err     error
+	orgErr  error
+}
+
+func (a *audienceStub) ResourceOrganization(context.Context, string, uuid.UUID) (uuid.UUID, error) {
+	return a.org, a.orgErr
+}
+
+func (a *audienceStub) CanAddressUser(context.Context, uuid.UUID, uuid.UUID) (bool, error) {
+	return a.allowed, a.err
+}
+
+func TestOwnerEventAudience(t *testing.T) {
+	org, user, campaign := uuid.New(), uuid.New(), uuid.New()
+	for _, state := range []string{"unrestricted", "restricted", "revoked", "error", "missing", "resource error", "wrong org"} {
+		t.Run(state, func(t *testing.T) {
+			resolver := &audienceStub{org: org, allowed: true}
+			var audience AudienceResolver = resolver
+			switch state {
+			case "restricted", "revoked":
+				resolver.allowed = false
+			case "error":
+				resolver.err = errors.New("unavailable")
+			case "missing":
+				audience = nil
+			case "resource error":
+				resolver.orgErr = errors.New("deleted resource")
+			case "wrong org":
+				resolver.org = uuid.New()
+			}
+			bus := &audienceBus{}
+			p := NewStreamingPublisher(bus, audience)
+			p.PublishCampaignEvent(context.Background(), &CampaignEvent{BaseEvent: BaseEvent{UserID: user.String(), EventType: EventCampaignUpdated}, CampaignID: campaign.String(), OrgID: org.String()})
+			if state == "wrong org" || state == "error" || state == "missing" || state == "resource error" {
+				if len(bus.events) != 0 {
+					t.Fatal("unresolved or mismatched resource audience was published")
+				}
+				return
+			}
+			if len(bus.events) != 1 || bus.events[0]["org_id"] != org.String() {
+				t.Fatalf("expected authorized organization route: %+v", bus.events)
+			}
+			wantUser := state == "unrestricted"
+			if (bus.events[0]["user_id"] == user.String()) != wantUser || (bus.attrs[0]["user_id"] == user.String()) != wantUser {
+				t.Fatalf("owner shortcut differs from membership: event=%+v attrs=%+v", bus.events, bus.attrs)
+			}
+		})
+	}
+}
+
+func TestResourceOnlyOwnerEvents(t *testing.T) {
+	org, user, mailbox, task := uuid.New(), uuid.New(), uuid.New(), uuid.New()
+	resolver := &audienceStub{org: org}
+	bus := &audienceBus{}
+	p := NewStreamingPublisher(bus, resolver)
+	p.PublishEmailError(context.Background(), user.String(), mailbox, uuid.Nil, "private", "private")
+	p.PublishTaskStatus(context.Background(), user.String(), task, EventTaskCompleted, "private", nil)
+	p.PublishWarmupStats(context.Background(), user.String(), mailbox, nil)
+	if len(bus.events) != 3 {
+		t.Fatalf("resource resolution lost events: %+v", bus.events)
+	}
+	for _, event := range bus.events {
+		if event["org_id"] != org.String() || event["user_id"] != nil {
+			t.Fatalf("restricted owner received user route: %+v", event)
+		}
+	}
+	resolver.orgErr = errors.New("unavailable")
+	p.PublishEmailError(context.Background(), user.String(), mailbox, uuid.Nil, "private", "private")
+	NewStreamingPublisher(bus).PublishEmailError(context.Background(), user.String(), mailbox, uuid.Nil, "private", "private")
+	if len(bus.events) != 3 {
+		t.Fatal("unresolved events must fail closed")
+	}
+}
+
+func TestPersonalAndWorkspaceNotificationAudience(t *testing.T) {
+	ctx := context.Background()
+	user, org := uuid.New(), uuid.New()
+	bus := &audienceBus{}
+	resolver := &audienceStub{org: org, allowed: true}
+	p := NewStreamingPublisher(bus, resolver)
+	p.PublishNotificationCreated(ctx, user.String(), uuid.NewString(), "campaign_completed", "private", "", &org)
+	if len(bus.events) != 1 || bus.events[0]["org_id"] != nil || bus.events[0]["user_id"] != user.String() {
+		t.Fatalf("recipient notice must never broadcast to workspace: %+v", bus.events)
+	}
+	resolver.allowed = false
+	p.PublishNotificationCreated(ctx, user.String(), uuid.NewString(), "campaign_completed", "private", "", &org)
+	NewStreamingPublisher(bus).PublishNotificationCreated(ctx, user.String(), uuid.NewString(), "campaign_completed", "private", "", &org)
+	NewStreamingPublisher(bus).PublishNotificationCreated(ctx, user.String(), uuid.NewString(), "campaign_completed", "private", "")
+	if len(bus.events) != 1 {
+		t.Fatal("restricted/unresolved notice published")
+	}
+	NewStreamingPublisher(bus).PublishNotificationCreated(ctx, user.String(), uuid.NewString(), "security_new_signin", "sign-in", "")
+	NewStreamingPublisher(bus).PublishSessionsRevoked(ctx, user)
+	if len(bus.events) != 3 || bus.events[1]["user_id"] != user.String() || bus.events[2]["user_id"] != user.String() {
+		t.Fatalf("personal events must survive without resolver: %+v", bus.events)
+	}
+}

internal/infrastructure/pubsub/client.go

@@ -132,14 +132,17 @@ type eventBus interface {
 
 // StreamingPublisher handles real-time streaming to users
 type StreamingPublisher struct {
-	client eventBus
+	client   eventBus
+	audience AudienceResolver
 }
 
 // NewStreamingPublisher creates a new streaming publisher
-func NewStreamingPublisher(client eventBus) *StreamingPublisher {
-	return &StreamingPublisher{
-		client: client,
+func NewStreamingPublisher(client eventBus, audience ...AudienceResolver) *StreamingPublisher {
+	p := &StreamingPublisher{client: client}
+	if len(audience) > 0 {
+		p.audience = audience[0]
 	}
+	return p
 }
 
 // Topic names for real-time updates
@@ -196,7 +199,7 @@ func (p *StreamingPublisher) PublishTaskStatus(ctx context.Context, userID strin
 		"event_type": string(eventType),
 	}
 
-	if err := p.client.Publish(ctx, TopicTaskStatus, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicTaskStatus, event, attrs); err != nil {
 		errs.CaptureException(fmt.Errorf("failed to publish task status: %w", err))
 	}
 }
@@ -226,7 +229,7 @@ func (p *StreamingPublisher) PublishEmailError(ctx context.Context, userID strin
 		"event_type": string(EventError),
 	}
 
-	if err := p.client.Publish(ctx, TopicEmailError, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicEmailError, event, attrs); err != nil {
 		errs.CaptureException(fmt.Errorf("failed to publish email error: %w", err))
 	}
 }
@@ -255,7 +258,7 @@ func (p *StreamingPublisher) PublishEmailWarning(ctx context.Context, userID str
 		"event_type": string(EventWarning),
 	}
 
-	if err := p.client.Publish(ctx, TopicEmailWarning, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicEmailWarning, event, attrs); err != nil {
 		errs.CaptureException(fmt.Errorf("failed to publish email warning: %w", err))
 	}
 }
@@ -280,7 +283,7 @@ func (p *StreamingPublisher) PublishCampaignProgress(ctx context.Context, userID
 		"event_type":  "CAMPAIGN_PROGRESS",
 	}
 
-	if err := p.client.Publish(ctx, TopicCampaignUpdate, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicCampaignUpdate, event, attrs); err != nil {
 		errs.CaptureException(fmt.Errorf("failed to publish campaign progress: %w", err))
 	}
 }
@@ -306,7 +309,7 @@ func (p *StreamingPublisher) PublishWarmupStats(ctx context.Context, userID stri
 		"event_type": "WARMUP_STATS",
 	}
 
-	if err := p.client.Publish(ctx, TopicWarmupUpdate, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicWarmupUpdate, event, attrs); err != nil {
 		errs.CaptureException(fmt.Errorf("failed to publish warmup stats: %w", err))
 	}
 }

internal/infrastructure/pubsub/events.go

@@ -282,7 +282,7 @@ func (p *StreamingPublisher) PublishPageHit(ctx context.Context, event *PageHitE
 		"contact_id": event.ContactID,
 		"event_type": string(event.EventType),
 	}
-	_ = p.client.Publish(ctx, TopicCampaignUpdate, event, attrs)
+	_ = p.publish(ctx, TopicCampaignUpdate, event, attrs)
 }
 
 // TaskProgressEvent for detailed campaign task progress
@@ -334,7 +334,7 @@ func (p *StreamingPublisher) PublishMeeting(ctx context.Context, userID string,
 		"user_id":    userID,
 		"event_type": string(eventType),
 	}
-	if err := p.client.Publish(ctx, TopicUserEvents, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicUserEvents, event, attrs); err != nil {
 		// Best-effort: realtime is a nicety, not a requirement.
 	}
 }
@@ -356,7 +356,7 @@ func (p *StreamingPublisher) PublishEmailReceived(ctx context.Context, event *Em
 		"event_type": string(EventEmailReceived),
 	}
 
-	if err := p.client.Publish(ctx, TopicEmailInbox, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicEmailInbox, event, attrs); err != nil {
 		// Log error but don't fail
 	}
 }
@@ -376,7 +376,7 @@ func (p *StreamingPublisher) PublishEmailUpdated(ctx context.Context, event *Ema
 		"event_type": string(EventEmailUpdated),
 	}
 
-	if err := p.client.Publish(ctx, TopicEmailInbox, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicEmailInbox, event, attrs); err != nil {
 		// Log error but don't fail
 	}
 }
@@ -396,7 +396,7 @@ func (p *StreamingPublisher) PublishEmailDeleted(ctx context.Context, event *Ema
 		"event_type": string(EventEmailDeleted),
 	}
 
-	if err := p.client.Publish(ctx, TopicEmailInbox, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicEmailInbox, event, attrs); err != nil {
 		// Log error but don't fail
 	}
 }
@@ -422,7 +422,7 @@ func (p *StreamingPublisher) PublishContactsReload(ctx context.Context, userID,
 		"event_type": string(EventContactsReload),
 	}
 
-	if err := p.client.Publish(ctx, TopicBulkOps, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicBulkOps, event, attrs); err != nil {
 		// Log error but don't fail
 	}
 }
@@ -446,7 +446,7 @@ func (p *StreamingPublisher) PublishOrgContactsReload(ctx context.Context, orgID
 		"org_id":     orgID,
 		"event_type": string(EventContactsReload),
 	}
-	_ = p.client.Publish(ctx, TopicBulkOps, event, attrs)
+	_ = p.publish(ctx, TopicBulkOps, event, attrs)
 }
 
 // PublishBulkProgress sends bulk operation progress update
@@ -463,7 +463,7 @@ func (p *StreamingPublisher) PublishBulkProgress(ctx context.Context, event *Bul
 		"event_type":   string(event.EventType),
 	}
 
-	if err := p.client.Publish(ctx, TopicBulkOps, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicBulkOps, event, attrs); err != nil {
 		// Log error but don't fail
 	}
 }
@@ -481,7 +481,7 @@ func (p *StreamingPublisher) PublishCampaignEvent(ctx context.Context, event *Ca
 		"campaign_id": event.CampaignID,
 		"event_type":  string(event.EventType),
 	}
-	if err := p.client.Publish(ctx, TopicCampaignUpdate, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicCampaignUpdate, event, attrs); err != nil {
 		// Log error but don't fail
 	}
 }
@@ -505,7 +505,7 @@ func (p *StreamingPublisher) PublishAccountEvent(ctx context.Context, event *Acc
 		topicID = TopicEmailError
 	}
 
-	if err := p.client.Publish(ctx, topicID, event, attrs); err != nil {
+	if err := p.publish(ctx, topicID, event, attrs); err != nil {
 		// Log error but don't fail
 	}
 }
@@ -588,7 +588,7 @@ func (p *StreamingPublisher) PublishAuditCreated(ctx context.Context, orgID, act
 		"event_type": string(EventAuditCreated),
 	}
 
-	if err := p.client.Publish(ctx, TopicUserEvents, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicUserEvents, event, attrs); err != nil {
 		// Best-effort: realtime is a nicety, not a requirement.
 	}
 }
@@ -623,7 +623,7 @@ func (p *StreamingPublisher) PublishFormSubmission(ctx context.Context, orgID, f
 		"org_id":     orgID.String(),
 		"event_type": string(EventFormSubmission),
 	}
-	if err := p.client.Publish(ctx, TopicUserEvents, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicUserEvents, event, attrs); err != nil {
 		// Best-effort: realtime is a nicety, not a requirement.
 	}
 }
@@ -657,7 +657,7 @@ func (p *StreamingPublisher) PublishPlacementTest(ctx context.Context, orgID, te
 		"org_id":     orgID.String(),
 		"event_type": string(EventPlacementTest),
 	}
-	_ = p.client.Publish(ctx, TopicUserEvents, event, attrs)
+	_ = p.publish(ctx, TopicUserEvents, event, attrs)
 }
 
 // PublishPlacementBatch emits the placement signal for a batch that moved
@@ -676,7 +676,7 @@ func (p *StreamingPublisher) PublishPlacementBatch(ctx context.Context, orgID, b
 		"org_id":     orgID.String(),
 		"event_type": string(EventPlacementTest),
 	}
-	_ = p.client.Publish(ctx, TopicUserEvents, event, attrs)
+	_ = p.publish(ctx, TopicUserEvents, event, attrs)
 }
 
 // AutomationEvent is an org-scoped automation lifecycle/run signal. The web
@@ -723,7 +723,7 @@ func (p *StreamingPublisher) PublishMailboxImportProgress(ctx context.Context, o
 		Status:    status,
 	}
 	attrs := map[string]string{"org_id": orgID.String(), "event_type": string(EventMailboxImportProgress)}
-	_ = p.client.Publish(ctx, TopicUserEvents, event, attrs)
+	_ = p.publish(ctx, TopicUserEvents, event, attrs)
 }
 
 // ContactImportEvent is the org-scoped payload for CONTACT_IMPORT_PROGRESS.
@@ -747,7 +747,7 @@ func (p *StreamingPublisher) PublishContactImportProgress(ctx context.Context, o
 		Status:    status,
 	}
 	attrs := map[string]string{"org_id": orgID.String(), "event_type": string(EventContactImportProgress)}
-	_ = p.client.Publish(ctx, TopicUserEvents, event, attrs)
+	_ = p.publish(ctx, TopicUserEvents, event, attrs)
 }
 
 // CRMSyncedEvent names what changed; ids only, the records stay behind their
@@ -771,7 +771,7 @@ func (p *StreamingPublisher) PublishCRMSynced(ctx context.Context, orgID uuid.UU
 		ContactID: contactID,
 	}
 	attrs := map[string]string{"org_id": orgID.String(), "event_type": string(EventCRMSynced)}
-	_ = p.client.Publish(ctx, TopicUserEvents, event, attrs)
+	_ = p.publish(ctx, TopicUserEvents, event, attrs)
 }
 
 // PublishAIResearchProgress emits AI_RESEARCH_PROGRESS for one completed run.
@@ -795,7 +795,7 @@ func (p *StreamingPublisher) PublishAIResearchProgress(ctx context.Context, orgI
 		"org_id":     orgID.String(),
 		"event_type": string(EventAIResearchProgress),
 	}
-	if err := p.client.Publish(ctx, TopicUserEvents, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicUserEvents, event, attrs); err != nil {
 		// Best-effort: realtime is a nicety, not a requirement.
 	}
 }
@@ -833,7 +833,7 @@ func (p *StreamingPublisher) PublishAIDraftReady(ctx context.Context, orgID, act
 		"org_id":     orgID.String(),
 		"event_type": string(EventAIDraftReady),
 	}
-	if err := p.client.Publish(ctx, TopicUserEvents, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicUserEvents, event, attrs); err != nil {
 		// Best-effort: realtime is a nicety, not a requirement.
 	}
 }
@@ -868,7 +868,7 @@ func (p *StreamingPublisher) PublishCreditsLow(ctx context.Context, orgID uuid.U
 		"org_id":     orgID.String(),
 		"event_type": string(EventBillingCreditsLow),
 	}
-	if err := p.client.Publish(ctx, TopicUserEvents, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicUserEvents, event, attrs); err != nil {
 		// Best-effort: realtime is a nicety, not a requirement.
 	}
 }
@@ -901,7 +901,7 @@ func (p *StreamingPublisher) PublishCreditsChanged(ctx context.Context, orgID uu
 		"org_id":     orgID.String(),
 		"event_type": string(EventBillingCreditsChanged),
 	}
-	if err := p.client.Publish(ctx, TopicUserEvents, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicUserEvents, event, attrs); err != nil {
 		// Best-effort: realtime is a nicety, not a requirement.
 	}
 }
@@ -925,7 +925,7 @@ func (p *StreamingPublisher) PublishAutomationEvent(ctx context.Context, orgID,
 		"org_id":     orgID.String(),
 		"event_type": string(eventType),
 	}
-	if err := p.client.Publish(ctx, TopicUserEvents, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicUserEvents, event, attrs); err != nil {
 		// Best-effort: realtime is a nicety, not a requirement.
 	}
 }
@@ -968,7 +968,7 @@ func (p *StreamingPublisher) PublishCustomEvent(ctx context.Context, orgID, acto
 		"org_id":     orgID.String(),
 		"event_type": string(EventCustomFired),
 	}
-	if err := p.client.Publish(ctx, TopicUserEvents, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicUserEvents, event, attrs); err != nil {
 		// Best-effort: realtime is a nicety, not a requirement.
 	}
 }
@@ -1002,7 +1002,7 @@ func (p *StreamingPublisher) PublishPresencePolicy(ctx context.Context, orgID uu
 		"org_id":     orgID.String(),
 		"event_type": string(EventPresencePolicyUpdated),
 	}
-	if err := p.client.Publish(ctx, TopicUserEvents, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicUserEvents, event, attrs); err != nil {
 		// Best-effort: realtime is a nicety, not a requirement.
 	}
 }
@@ -1021,7 +1021,7 @@ func (p *StreamingPublisher) PublishSessionsRevoked(ctx context.Context, userID
 		"user_id":    userID.String(),
 		"event_type": string(EventSessionsRevoked),
 	}
-	if err := p.client.Publish(ctx, TopicUserEvents, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicUserEvents, event, attrs); err != nil {
 		// Best-effort: the sessions are already revoked, and every reconnect needs a live one.
 	}
 }
@@ -1036,7 +1036,7 @@ func (p *StreamingPublisher) PublishToUser(ctx context.Context, userID string, e
 		"user_id": userID,
 	}
 
-	if err := p.client.Publish(ctx, TopicUserEvents, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicUserEvents, event, attrs); err != nil {
 		// Log error but don't fail
 	}
 }
@@ -1045,6 +1045,7 @@ func (p *StreamingPublisher) PublishToUser(ctx context.Context, userID string, e
 // notification (the bell). Best-effort; the feed table is the source of truth.
 type NotificationEvent struct {
 	BaseEvent
+	OrgID          string `json:"org_id,omitempty"`
 	NotificationID string `json:"notification_id"`
 	Category       string `json:"category"`
 	Title          string `json:"title"`
@@ -1052,7 +1053,7 @@ type NotificationEvent struct {
 }
 
 // PublishNotificationCreated pushes a new-notification event to a single user.
-func (p *StreamingPublisher) PublishNotificationCreated(ctx context.Context, userID, notifID, category, title, link string) {
+func (p *StreamingPublisher) PublishNotificationCreated(ctx context.Context, userID, notifID, category, title, link string, orgID ...*uuid.UUID) {
 	if p == nil || p.client == nil || userID == "" {
 		return
 	}
@@ -1067,11 +1068,14 @@ func (p *StreamingPublisher) PublishNotificationCreated(ctx context.Context, use
 		Title:          title,
 		Link:           link,
 	}
+	if len(orgID) > 0 && orgID[0] != nil {
+		event.OrgID = orgID[0].String()
+	}
 	attrs := map[string]string{
 		"user_id":    userID,
 		"event_type": string(EventNotificationCreated),
 	}
-	if err := p.client.Publish(ctx, TopicUserEvents, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicUserEvents, event, attrs); err != nil {
 		// Best-effort: realtime is a nicety, not a requirement.
 	}
 }
@@ -1093,7 +1097,7 @@ func (p *StreamingPublisher) PublishTrackingEvent(ctx context.Context, event *Tr
 		attrs["org_id"] = event.OrgID
 	}
 
-	if err := p.client.Publish(ctx, TopicCampaignUpdate, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicCampaignUpdate, event, attrs); err != nil {
 		// Log error but don't fail
 	}
 }
@@ -1114,7 +1118,7 @@ func (p *StreamingPublisher) PublishTaskProgress(ctx context.Context, event *Tas
 		"event_type":  string(EventTaskProgress),
 	}
 
-	if err := p.client.Publish(ctx, TopicCampaignUpdate, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicCampaignUpdate, event, attrs); err != nil {
 		// Log error but don't fail
 	}
 }
@@ -1156,7 +1160,7 @@ func (p *StreamingPublisher) PublishEmailSent(ctx context.Context, event *TaskPr
 		"event_type":  string(EventEmailSent),
 	}
 
-	if err := p.client.Publish(ctx, TopicCampaignUpdate, event, attrs); err != nil {
+	if err := p.publish(ctx, TopicCampaignUpdate, event, attrs); err != nil {
 		// Best-effort: realtime is a nicety, not a requirement.
 	}
 }

internal/repository/notification_audience.go

@@ -0,0 +1,11 @@
+package repository
+
+// Workspace-wide notices require a current accepted unrestricted membership, including stored rows.
+const notificationRecipientVisibleSQL = `(
+	(n.organization_id IS NULL AND n.category = 'security_new_signin')
+	OR (n.organization_id IS NOT NULL AND EXISTS (
+		SELECT 1 FROM organization_members om
+		WHERE om.organization_id = n.organization_id AND om.user_id = n.user_id
+			AND om.accepted_at IS NOT NULL AND (om.role = 'owner' OR om.access_scope = 'workspace')
+	))
+)`

internal/repository/notification_audience_live_test.go

@@ -0,0 +1,92 @@
+package repository
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+func TestLiveNotificationAudienceRevocation(t *testing.T) {
+	for _, revoke := range []string{"restricted", "removed", "unaccepted"} {
+		t.Run(revoke, func(t *testing.T) {
+			_, pool := liveContactDB(t)
+			f := newSharedOrgFixture(t, pool)
+			ctx := context.Background()
+			repo := NewNotificationRepository(pool)
+			due := time.Now().Add(-time.Minute)
+			var noticeIDs []uuid.UUID
+			for _, category := range []models.NotificationCategory{models.NotifCampaignPaused, models.NotifSecuritySignIn} {
+				var org *uuid.UUID
+				if category == models.NotifCampaignPaused {
+					org = &f.org
+				}
+				n, err := repo.Create(ctx, &models.Notification{UserID: f.mate, OrganizationID: org, Category: category, Title: "private", EmailState: "pending", EmailDueAt: &due})
+				if err != nil {
+					t.Fatal(err)
+				}
+				noticeIDs = append(noticeIDs, n.ID)
+			}
+			for _, unread := range []bool{false, true} {
+				rows, err := repo.List(ctx, f.mate, 50, unread)
+				if err != nil || len(rows) != 2 {
+					t.Fatalf("unrestricted feed: len=%d err=%v", len(rows), err)
+				}
+			}
+			if n, err := repo.CountUnread(ctx, f.mate); err != nil || n != 2 {
+				t.Fatalf("unrestricted count=%d err=%v", n, err)
+			}
+			// The resource creator is a member, not an authorization shortcut.
+			if _, err := pool.Exec(ctx, `UPDATE campaigns SET user_id=$1 WHERE id=$2`, f.mate, f.campaign); err != nil {
+				t.Fatal(err)
+			}
+			audience := NewEventAudienceRepository(pool)
+			org, err := audience.ResourceOrganization(ctx, "campaign", f.campaign)
+			if err != nil || org != f.org {
+				t.Fatalf("resource org=%v err=%v", org, err)
+			}
+			if allowed, err := audience.CanAddressUser(ctx, f.org, f.mate); err != nil || !allowed {
+				t.Fatalf("unrestricted owner=%v err=%v", allowed, err)
+			}
+			query := `UPDATE organization_members SET access_scope='restricted' WHERE organization_id=$1 AND user_id=$2`
+			if revoke == "removed" {
+				query = `DELETE FROM organization_members WHERE organization_id=$1 AND user_id=$2`
+			} else if revoke == "unaccepted" {
+				query = `UPDATE organization_members SET accepted_at=NULL WHERE organization_id=$1 AND user_id=$2`
+			}
+			if _, err := pool.Exec(ctx, query, f.org, f.mate); err != nil {
+				t.Fatal(err)
+			}
+			if allowed, err := audience.CanAddressUser(ctx, f.org, f.mate); err != nil || allowed {
+				t.Fatalf("revoked creator=%v err=%v", allowed, err)
+			}
+			for _, unread := range []bool{false, true} {
+				rows, err := repo.List(ctx, f.mate, 50, unread)
+				if err != nil || len(rows) != 1 || rows[0].ID != noticeIDs[1] {
+					t.Fatalf("revoked feed: %+v err=%v", rows, err)
+				}
+			}
+			if n, err := repo.CountUnread(ctx, f.mate); err != nil || n != 1 {
+				t.Fatalf("revoked count=%d err=%v", n, err)
+			}
+			claimed, err := repo.ClaimDueEmails(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+			for _, n := range claimed {
+				if n.ID == noticeIDs[0] {
+					t.Fatal("revoked content claimed for deferred email")
+				}
+			}
+			var state string
+			if err := pool.QueryRow(ctx, `SELECT email_state FROM notifications WHERE id=$1`, noticeIDs[0]).Scan(&state); err != nil || state != "skipped" {
+				t.Fatalf("revoked email_state=%q err=%v", state, err)
+			}
+			if _, err := repo.Create(ctx, &models.Notification{UserID: f.mate, OrganizationID: &f.org, Category: models.NotifCampaignPaused, Title: "denied"}); err == nil {
+				t.Fatal("revoked member persisted a new workspace notice")
+			}
+		})
+	}
+}

internal/repository/pg_event_audience.go

@@ -0,0 +1,48 @@
+package repository
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+type EventAudienceRepository struct {
+	db *pgxpool.Pool
+}
+
+func NewEventAudienceRepository(db *pgxpool.Pool) *EventAudienceRepository {
+	return &EventAudienceRepository{db: db}
+}
+
+func (r *EventAudienceRepository) ResourceOrganization(ctx context.Context, kind string, id uuid.UUID) (uuid.UUID, error) {
+	var query string
+	switch kind {
+	case "campaign":
+		query = `SELECT organization_id FROM campaigns WHERE id = $1`
+	case "mailbox":
+		query = `SELECT organization_id FROM email_accounts WHERE id = $1`
+	case "task":
+		query = `SELECT ea.organization_id FROM tasks t JOIN email_accounts ea ON ea.id = t.email_account_id WHERE t.id = $1`
+	case "contact":
+		query = `SELECT organization_id FROM contacts WHERE id = $1`
+	case "booking":
+		query = `SELECT organization_id FROM meeting_bookings WHERE id = $1`
+	default:
+		return uuid.Nil, fmt.Errorf("unsupported event resource kind %q", kind)
+	}
+	var orgID uuid.UUID
+	err := r.db.QueryRow(ctx, query, id).Scan(&orgID)
+	return orgID, err
+}
+
+func (r *EventAudienceRepository) CanAddressUser(ctx context.Context, orgID, userID uuid.UUID) (bool, error) {
+	var allowed bool
+	err := r.db.QueryRow(ctx, `SELECT EXISTS (
+		SELECT 1 FROM organization_members
+		WHERE organization_id = $1 AND user_id = $2 AND accepted_at IS NOT NULL
+			AND (role = 'owner' OR access_scope = 'workspace')
+	)`, orgID, userID).Scan(&allowed)
+	return allowed, err
+}

internal/repository/pg_notification.go

@@ -113,7 +113,7 @@ func (r *notificationRepository) Create(ctx context.Context, n *models.Notificat
 	err := r.db.QueryRow(ctx, `
 		WITH msg AS (SELECT seen FROM unibox_emails WHERE id = $13 FOR SHARE),
 		seen AS (SELECT COALESCE((SELECT seen FROM msg), false) AS v),
-		candidate AS (SELECT $3::uuid AS organization_id, $4::text AS category,
+		candidate AS (SELECT $2::uuid AS user_id, $3::uuid AS organization_id, $4::text AS category,
 			$8::jsonb AS metadata, $13::uuid AS unibox_email_id)
 		INSERT INTO notifications (id, user_id, organization_id, category, title, body, link, metadata,
 			group_key, email_state, email_due_at, read_at, unibox_email_id)
@@ -122,7 +122,7 @@ func (r *notificationRepository) Create(ctx context.Context, n *models.Notificat
 			CASE WHEN seen.v AND $10 = 'pending' THEN NULL ELSE $11::timestamptz END,
 			CASE WHEN $12 OR seen.v THEN now() END,
 			$13
-		FROM seen, candidate n WHERE `+notificationReplyVisibleSQL+`
+		FROM seen, candidate n WHERE `+notificationReplyVisibleSQL+` AND `+notificationRecipientVisibleSQL+`
 		ON CONFLICT(id) DO UPDATE SET id=EXCLUDED.id
 		RETURNING created_at, (SELECT v FROM seen)`,
 		n.ID, n.UserID, n.OrganizationID, n.Category, n.Title, n.Body, n.Link, meta,
@@ -193,7 +193,7 @@ func (r *notificationRepository) List(ctx context.Context, userID uuid.UUID, lim
 		limit = 50
 	}
 	q := `SELECT id, user_id, organization_id, category, title, body, link, metadata, read_at, created_at
-		FROM notifications n WHERE user_id = $1 AND ` + notificationReplyVisibleSQL
+		FROM notifications n WHERE user_id = $1 AND ` + notificationReplyVisibleSQL + ` AND ` + notificationRecipientVisibleSQL
 	if unreadOnly {
 		q += ` AND read_at IS NULL`
 	}
@@ -220,7 +220,7 @@ func (r *notificationRepository) List(ctx context.Context, userID uuid.UUID, lim
 
 func (r *notificationRepository) CountUnread(ctx context.Context, userID uuid.UUID) (int, error) {
 	var c int
-	err := r.db.QueryRow(ctx, `SELECT COUNT(*) FROM notifications n WHERE user_id = $1 AND read_at IS NULL AND `+notificationReplyVisibleSQL, userID).Scan(&c)
+	err := r.db.QueryRow(ctx, `SELECT COUNT(*) FROM notifications n WHERE user_id = $1 AND read_at IS NULL AND `+notificationReplyVisibleSQL+` AND `+notificationRecipientVisibleSQL, userID).Scan(&c)
 	return c, err
 }
 
@@ -249,7 +249,7 @@ func (r *notificationRepository) MarkAllRead(ctx context.Context, userID uuid.UU
 // claims can be recovered back to pending.
 func (r *notificationRepository) ClaimDueEmails(ctx context.Context) ([]models.Notification, error) {
 	if _, err := r.db.Exec(ctx, `UPDATE notifications n SET email_state = 'skipped', email_due_at = NULL
-		WHERE email_state = 'pending' AND NOT `+notificationReplyVisibleSQL); err != nil {
+		WHERE email_state = 'pending' AND NOT (`+notificationReplyVisibleSQL+` AND `+notificationRecipientVisibleSQL+`)`); err != nil {
 		return nil, err
 	}
 	_, _ = r.db.Exec(ctx, `
@@ -260,7 +260,7 @@ func (r *notificationRepository) ClaimDueEmails(ctx context.Context) ([]models.N
 		UPDATE notifications SET email_state = 'sending', email_due_at = now()
 		WHERE id IN (
 			SELECT n.id FROM notifications n
-			WHERE n.email_state = 'pending' AND `+notificationReplyVisibleSQL+` AND (
+			WHERE n.email_state = 'pending' AND `+notificationReplyVisibleSQL+` AND `+notificationRecipientVisibleSQL+` AND (
 				n.user_id IN (
 					SELECT user_id FROM notifications
 					WHERE email_state = 'pending' AND email_due_at <= now())