From dbdd706f23fdbce2fe72072e6160ac87d1ff2fe1 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 10 Oct 2026 10:51:11 +0000 Subject: [PATCH] feat: enforce current workspace membership on owner-addressed realtime events and notification ingress, stored feeds, deferred email and push delivery --- cmd/backend/main.go | 4 +- cmd/consumer/main.go | 4 +- internal/app/consumer/sync_truth_test.go | 25 +++- internal/app/notification/audience_test.go | 107 ++++++++++++++ internal/app/notification/email.go | 21 +-- internal/app/notification/message_test.go | 16 ++- internal/app/notification/push.go | 20 ++- internal/app/notification/service.go | 33 ++++- internal/infrastructure/pubsub/audience.go | 114 +++++++++++++++ .../infrastructure/pubsub/audience_test.go | 130 ++++++++++++++++++ internal/infrastructure/pubsub/client.go | 21 +-- internal/infrastructure/pubsub/events.go | 66 ++++----- internal/repository/notification_audience.go | 11 ++ .../notification_audience_live_test.go | 92 +++++++++++++ internal/repository/pg_event_audience.go | 48 +++++++ internal/repository/pg_notification.go | 12 +- 16 files changed, 638 insertions(+), 86 deletions(-) create mode 100644 internal/app/notification/audience_test.go create mode 100644 internal/infrastructure/pubsub/audience.go create mode 100644 internal/infrastructure/pubsub/audience_test.go create mode 100644 internal/repository/notification_audience.go create mode 100644 internal/repository/notification_audience_live_test.go create mode 100644 internal/repository/pg_event_audience.go diff --git a/cmd/backend/main.go b/cmd/backend/main.go index 87d6ca6a8..c6cb9b166 100644 --- a/cmd/backend/main.go +++ b/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)") } diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index d00d402ca..62a9390a1 100644 --- a/cmd/consumer/main.go +++ b/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)") } diff --git a/internal/app/consumer/sync_truth_test.go b/internal/app/consumer/sync_truth_test.go index 6c0d1f6a8..761bb411d 100644 --- a/internal/app/consumer/sync_truth_test.go +++ b/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) diff --git a/internal/app/notification/audience_test.go b/internal/app/notification/audience_test.go new file mode 100644 index 000000000..1cac8ff4a --- /dev/null +++ b/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") + } +} diff --git a/internal/app/notification/email.go b/internal/app/notification/email.go index dc92d76ce..97ab07898 100644 --- a/internal/app/notification/email.go +++ b/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 { diff --git a/internal/app/notification/message_test.go b/internal/app/notification/message_test.go index 7830318cb..595a35950 100644 --- a/internal/app/notification/message_test.go +++ b/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) } diff --git a/internal/app/notification/push.go b/internal/app/notification/push.go index bc3643941..1bd072441 100644 --- a/internal/app/notification/push.go +++ b/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 { diff --git a/internal/app/notification/service.go b/internal/app/notification/service.go index 96beab31a..d86ebf7ba 100644 --- a/internal/app/notification/service.go +++ b/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 diff --git a/internal/infrastructure/pubsub/audience.go b/internal/infrastructure/pubsub/audience.go new file mode 100644 index 000000000..18e487c9d --- /dev/null +++ b/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) +} diff --git a/internal/infrastructure/pubsub/audience_test.go b/internal/infrastructure/pubsub/audience_test.go new file mode 100644 index 000000000..f7dc82178 --- /dev/null +++ b/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) + } +} diff --git a/internal/infrastructure/pubsub/client.go b/internal/infrastructure/pubsub/client.go index 7126b9d8e..c01564b0b 100644 --- a/internal/infrastructure/pubsub/client.go +++ b/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)) } } diff --git a/internal/infrastructure/pubsub/events.go b/internal/infrastructure/pubsub/events.go index be57d5463..d6f63c8e4 100644 --- a/internal/infrastructure/pubsub/events.go +++ b/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. } } diff --git a/internal/repository/notification_audience.go b/internal/repository/notification_audience.go new file mode 100644 index 000000000..797641fff --- /dev/null +++ b/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') + )) +)` diff --git a/internal/repository/notification_audience_live_test.go b/internal/repository/notification_audience_live_test.go new file mode 100644 index 000000000..1bc419c48 --- /dev/null +++ b/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") + } + }) + } +} diff --git a/internal/repository/pg_event_audience.go b/internal/repository/pg_event_audience.go new file mode 100644 index 000000000..d63fc5987 --- /dev/null +++ b/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 +} diff --git a/internal/repository/pg_notification.go b/internal/repository/pg_notification.go index c170a5c93..8598e73a7 100644 --- a/internal/repository/pg_notification.go +++ b/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())