From 66a105fb9fde48d44f1824ce787454489844290e Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Mon, 31 Aug 2026 08:45:27 -0700 Subject: [PATCH 1/9] feat: make segments a live campaign audience and round out lead management (issue #277): a campaign_segments link table with GET/PUT /campaigns/:id/segments (max 20, replace semantics), immediate enrolment of linked-segment members as leads plus targeted re-syncs on contact writes, segment edits, pin-ins and imports and a 2-minute backend sweep for drift, waking active campaigns and restarting completed ones through the full launch checks; a Segments manager dialog and toolbar button on the campaign Leads tab; Remove from campaign as the row action and a selection-bar bulk action so leads can leave a campaign without deleting the contact; campaign and segment pickers in the CSV import wizard backed by a new segment_ids import option that pins imported rows as include overrides; a delete guard naming linked campaigns; orgtransfer registration, live repo tests and docs for all of it --- cmd/backend/main.go | 15 + docs/content/docs/api/endpoints.mdx | 4 +- docs/content/docs/api/reference/campaigns.mdx | 73 +++++ docs/content/docs/api/reference/contacts.mdx | 5 +- docs/content/docs/guides/campaigns.mdx | 6 +- docs/content/docs/guides/contacts-crm.mdx | 4 +- docs/content/docs/guides/segments.mdx | 23 +- .../docs/guides/workspace-export-import.mdx | 2 +- internal/api/handler/segment.go | 50 ++++ internal/api/routes.go | 6 + internal/app/contact/handler.go | 3 + internal/app/contact/import.go | 56 ++++ internal/app/contact/service.go | 36 +++ internal/app/orgtransfer/spec.go | 6 + internal/app/segment/service.go | 193 +++++++++++- .../000114_campaign_segments.down.sql | 1 + .../000114_campaign_segments.up.sql | 11 + internal/models/contact_import.go | 5 +- internal/models/segment.go | 26 ++ internal/repository/pg_segment.go | 281 ++++++++++++++++-- internal/repository/segment_live_test.go | 72 +++++ .../components/app/contacts/ContactsTable.tsx | 103 ++++++- .../components/app/contacts/ImportWizard.tsx | 55 +++- ...sDialog.tsx => CampaignSegmentsDialog.tsx} | 103 +++++-- .../api/client/app/contacts/importContacts.ts | 1 + web/src/lib/api/client/app/segments/index.ts | 22 ++ web/src/lib/api/hooks/app/segments/index.ts | 26 ++ .../lib/api/models/app/segments/Segment.ts | 10 + 28 files changed, 1118 insertions(+), 80 deletions(-) create mode 100644 internal/infrastructure/db/migrations/000114_campaign_segments.down.sql create mode 100644 internal/infrastructure/db/migrations/000114_campaign_segments.up.sql rename web/src/components/app/segments/{AddSegmentLeadsDialog.tsx => CampaignSegmentsDialog.tsx} (70%) diff --git a/cmd/backend/main.go b/cmd/backend/main.go index 728dc8598..f436cbf59 100644 --- a/cmd/backend/main.go +++ b/cmd/backend/main.go @@ -1197,6 +1197,11 @@ func main() { contactService = contact.NewService(contactRepostory, subscriptionRepository, planRepository, streamingPublisher) segmentRepository := repository.NewSegmentRepository(primaryDB) segmentService = segment.NewService(segmentRepository, contactRepostory) + // Imports write segment include overrides, and every contact-mutating + // path re-enrols the org's segment-linked campaigns. + if aware, ok := contactService.(contact.SegmentAware); ok { + aware.WireSegments(segmentRepository, segmentService) + } // A visibly bad import is filed on the workspace's posture. On its own // it can only reach `watch`, which changes nothing. if aware, ok := contactService.(contact.OrgRiskAware); ok && orgRiskService != nil { @@ -1290,6 +1295,9 @@ func main() { // and Cloud Tasks client exist. if segmentService != nil { segmentService.SetCampaignWaker(campaignService) + // A completed campaign whose linked segments grow is restarted + // through the full launch checks, never by a raw status flip. + segmentService.SetCampaignStarter(campaignService) } if contactService != nil { contactService.SetCampaignWaker(campaignService) @@ -1582,6 +1590,13 @@ func main() { // started, so without this a stranded campaign stops sending forever. go tasksService.StartCampaignReconciler(ctx, 5*time.Minute) + // Segment-linked campaigns: enrol contacts that drifted into a linked + // segment (date windows, engagement counters, nested segments) that + // the write-path syncs cannot see. + if segmentService != nil { + go segmentService.StartCampaignSegmentSync(ctx, 2*time.Minute) + } + // Worker reconciler: assign + (re)load every active mailbox onto its // worker. Workers hold accounts in memory only, so this is what makes a // freshly onboarded account send/sync and re-seeds a restarted worker. diff --git a/docs/content/docs/api/endpoints.mdx b/docs/content/docs/api/endpoints.mdx index f8652f3f6..c952bd220 100644 --- a/docs/content/docs/api/endpoints.mdx +++ b/docs/content/docs/api/endpoints.mdx @@ -48,6 +48,8 @@ All paths below are relative to the versioned base URL `https://api.warmbly.com/ | PATCH | `/campaigns/:id` | `WRITE_CAMPAIGNS` | | DELETE | `/campaigns/:id` | `WRITE_CAMPAIGNS` | | POST | `/campaigns/:id/duplicate` | `WRITE_CAMPAIGNS` | +| GET | `/campaigns/:id/segments` | `READ_CAMPAIGNS` | +| PUT | `/campaigns/:id/segments` | `WRITE_CAMPAIGNS` | | GET | `/campaigns/:id/advanced` | `READ_CAMPAIGNS` | | PATCH | `/campaigns/:id/advanced` | `WRITE_CAMPAIGNS` | | GET | `/campaigns/:id/ab-variants` | `READ_CAMPAIGNS` | @@ -120,7 +122,7 @@ AI contact research charges credits (2 per run, billable even when it finds noth | GET | `/segments/:id/overrides` | `READ_CONTACTS` | | POST | `/segments/:id/add-to-campaign` | `WRITE_CAMPAIGNS` | -Segments are saved contact audiences: a condition list plus per-contact manual overrides, evaluated live. Contact scopes cover them because a segment is a view over contacts; enrolling one into a campaign writes leads, so that call takes the campaign write scope. `POST /contacts/search` and `POST /contacts/export` accept `segment_ids` to scope any contact query to a segment. See [contacts](/api/reference/contacts/#segments) for the condition format. +Segments are saved contact audiences: a condition list plus per-contact manual overrides, evaluated live. Contact scopes cover them because a segment is a view over contacts; enrolling one into a campaign writes leads, so that call takes the campaign write scope, as does managing a campaign's linked segments (`/campaigns/:id/segments` above). `POST /contacts/search` and `POST /contacts/export` accept `segment_ids` to scope any contact query to a segment. See [contacts](/api/reference/contacts/#segments) for the condition format. ### Lead sync diff --git a/docs/content/docs/api/reference/campaigns.mdx b/docs/content/docs/api/reference/campaigns.mdx index f4a4dbe1f..a7cb7690c 100644 --- a/docs/content/docs/api/reference/campaigns.mdx +++ b/docs/content/docs/api/reference/campaigns.mdx @@ -775,6 +775,79 @@ Atomically replace the campaign's explicit sender pool with the supplied list. * A `data` array of the resulting sender objects (same shape as [list senders](#list-campaign-senders)). +## List linked segments + +`GET /campaigns/:id/segments` + +Return the segments linked to the campaign as live audience sources. **Scope** `READ_CAMPAIGNS` · **Org permission** `view_campaigns`. + +| Parameter | In | Type | Description | +| --- | --- | --- | --- | +| `id` | path | uuid | Campaign id. | + +### Response + +A `data` array of link objects (no pagination wrapper). + +```json +{ + "data": [ + { + "segment_id": "4b9e...", + "name": "Warm leads", + "color": "#0ea5e9", + "description": "Replied or clicked in the last 30 days", + "contact_count": 412, + "linked_at": "2026-06-10T12:00:00Z" + } + ] +} +``` + +## Replace linked segments + +`PUT /campaigns/:id/segments` + +Atomically replace the campaign's linked segments with the supplied set (up to 20). Every current member of a newly linked segment is enrolled as a lead immediately, and contacts who enter a linked segment later are enrolled automatically, within about 2 minutes. Enrolment is additive: a contact who leaves a segment keeps their lead row, and removing a segment from the set stops future enrolment without touching existing leads. An active campaign is woken to send to the new leads; a completed campaign is restarted through the full launch checks when a linked segment grows. A linked segment cannot be deleted (`DELETE /segments/:id` returns `409`) until it is removed here. **Scope** `WRITE_CAMPAIGNS` · **Org permission** `manage_campaigns`. + +| Parameter | In | Type | Description | +| --- | --- | --- | --- | +| `id` | path | uuid | Campaign id. | + +### Request body + +| Field | Type | Required | Description | +| --- | --- | --- | --- | +| `segment_ids` | uuid[] | yes | The full new set of linked segments, max 20. An empty array detaches them all. | + +```json +{ + "segment_ids": ["4b9e...", "7c2f..."] +} +``` + +### Response + +The resulting links plus how many leads the call enrolled. + +```json +{ + "data": [ + { + "segment_id": "4b9e...", + "name": "Warm leads", + "color": "#0ea5e9", + "description": "Replied or clicked in the last 30 days", + "contact_count": 412, + "linked_at": "2026-06-10T12:00:00Z" + } + ], + "added": 397 +} +``` + +For a one-time snapshot enrolment instead of a live link, see [add a segment to a campaign](/api/reference/contacts/#add-to-a-campaign). + ## Verify campaign tracking domain `POST /campaigns/:id/tracking-domain/verify` diff --git a/docs/content/docs/api/reference/contacts.mdx b/docs/content/docs/api/reference/contacts.mdx index a72fae150..1053bdbab 100644 --- a/docs/content/docs/api/reference/contacts.mdx +++ b/docs/content/docs/api/reference/contacts.mdx @@ -366,6 +366,7 @@ Send `multipart/form-data` with a `file` field and an `options` field containing | `has_header` | boolean | Yes | Whether the first row is a header. | | `category_ids` | string[] | No | Categories to assign to imported contacts. | | `campaign_ids` | string[] | No | Campaigns to add imported contacts to. | +| `segment_ids` | string[] | No | Segments to pin imported contacts into, as a manual include override. Applies to imported, updated, and skipped-but-linked contacts alike. | | `subscribed_default` | boolean | No | Subscription state for new contacts when no subscribed column is mapped. Defaults to true. | ```json @@ -1028,7 +1029,7 @@ Engagement counters add up every campaign the contact has been in, and opens cou ### List, create, read, update, delete -`GET /segments` returns every segment with live counts under `data`. `POST /segments` creates one from `name` (required), `description`, `color` (`#rrggbb`), `match` and `conditions`; a duplicate name is a `409`. `GET /segments/:id` returns one segment. `PATCH /segments/:id` accepts the same fields, all optional. `DELETE /segments/:id` returns `204`, or `409` when another segment's conditions reference it. +`GET /segments` returns every segment with live counts under `data`. `POST /segments` creates one from `name` (required), `description`, `color` (`#rrggbb`), `match` and `conditions`; a duplicate name is a `409`. `GET /segments/:id` returns one segment. `PATCH /segments/:id` accepts the same fields, all optional. `DELETE /segments/:id` returns `204`, or `409` when another segment's conditions reference it or a campaign has it [linked as a live audience](/api/reference/campaigns/#replace-linked-segments); detach it first. Auth: **Scope** `READ_CONTACTS` for reads, `WRITE_CONTACTS` for writes · **Org permission** `view_contacts` / `manage_contacts` @@ -1068,6 +1069,6 @@ Sequence action steps `add_to_segment` and `remove_from_segment` take a `segment { "campaign_id": "…" } ``` -Enrols every current member as a lead. Contacts already in the campaign are skipped, each new lead gets a `campaign_added` activity, and a running campaign is woken so the leads are scheduled. Returns `{ "campaign_id", "added", "members" }`. This is a snapshot: later members are not added until the call is repeated. Safe to retry. +Enrols every current member as a lead. Contacts already in the campaign are skipped, each new lead gets a `campaign_added` activity, and a running campaign is woken so the leads are scheduled. Returns `{ "campaign_id", "added", "members" }`. This is a snapshot: later members are not added until the call is repeated. Safe to retry. For a live link that keeps enrolling members as they join the segment, see [linked segments](/api/reference/campaigns/#list-linked-segments). Auth: **Scope** `WRITE_CAMPAIGNS` · **Org permission** `manage_campaigns` diff --git a/docs/content/docs/guides/campaigns.mdx b/docs/content/docs/guides/campaigns.mdx index a8ba120de..815cc08b0 100644 --- a/docs/content/docs/guides/campaigns.mdx +++ b/docs/content/docs/guides/campaigns.mdx @@ -59,7 +59,11 @@ Anything above `50`/day per cold mailbox needs positive reputation signals and a ### Adding leads -The **Leads** tab takes contacts four ways. **From contacts** opens a picker over the workspace contact list: search by name, email or company, filter by category, tick individual people or **Select loaded**, or **Select all matching** to add everyone the search returns (up to 1,000 per batch; repeat for the rest). Contacts already in the campaign are marked as leads and skipped. **Import** runs the file import wizard with this campaign preselected, **Sheet sync** attaches a Google Sheet, and **Add lead** creates a single new contact in the campaign. Any workspace member with contact access can add leads to any campaign in the workspace, whoever created it. +The **Leads** tab takes contacts four ways. **From contacts** opens a picker over the workspace contact list: search by name, email or company, filter by category, tick individual people or **Select loaded**, or **Select all matching** to add everyone the search returns (up to 1,000 per batch; repeat for the rest). Contacts already in the campaign are marked as leads and skipped. **Import** runs the file import wizard with this campaign preselected (its Options step can also pin the file into segments), **Sheet sync** attaches a Google Sheet, and **Add lead** creates a single new contact in the campaign. Any workspace member with contact access can add leads to any campaign in the workspace, whoever created it. + +**Segments** on the same toolbar links [segments](/guides/segments/) to the campaign as a live audience, up to 20 per campaign. Linking enrols every current member immediately, and contacts who enter a linked segment later are enrolled on their own, within a couple of minutes. An active campaign wakes to send to them, a finished one restarts through the usual launch checks, and a paused one accumulates them for later. Enrolment is additive: a contact who leaves the segment keeps their lead row, and detaching a segment stops future enrolment without touching existing leads. See [linking a segment to a campaign](/guides/segments/#linking-a-segment-to-a-campaign). + +A lead can also be taken out again: the remove button on a lead's row, or **Remove from campaign** in the selection bar when several are ticked, drops the lead from this campaign without deleting the contact. ### Lead statuses diff --git a/docs/content/docs/guides/contacts-crm.mdx b/docs/content/docs/guides/contacts-crm.mdx index a5f702c7d..c9ef90347 100644 --- a/docs/content/docs/guides/contacts-crm.mdx +++ b/docs/content/docs/guides/contacts-crm.mdx @@ -24,13 +24,15 @@ You must map at least one column to **Email**. `Dana@Acme.com` and `dana@acme.com` are the same contact. Choose **Skip existing** (leave untouched), **Update existing** (merge new values in), or **Create duplicates** (force a new one, falling back to update if a uniqueness rule blocks it). -Skipping still adds the contact to the campaign and categories the import targets: it leaves their fields alone, it does not leave them out of the list. If the same address appears twice in one file it becomes one contact, and the extra rows count as skipped. +Skipping still adds the contact to the campaigns, segments, and categories the import targets: it leaves their fields alone, it does not leave them out of the list. If the same address appears twice in one file it becomes one contact, and the extra rows count as skipped. Rows with a missing or invalid email are reported with a line number and reason rather than imported. A blank cell never erases a value you already have, so re-importing a partial export enriches contacts instead of wiping them. Problems with the mapping itself, an unnamed custom field or a name Warmbly cannot use, are reported once before anything is written, so a single typo can never fail every row. +The **Options** step also decides where the contacts land: **Add to campaigns** enrols everyone in the file as leads of the campaigns you pick (hidden when the wizard was started from a campaign's Leads tab, since that campaign is already the target), and **Add to segments** pins them into the [segments](/guides/segments/) you pick as manual includes, so they stay members whatever the segment's conditions say. + **From Google Sheets**, a sheet is a reusable **sync source** rather than a one-time upload. Connect it once (Warmbly reads only the tab you choose and never writes back), paste the spreadsheet ID from the URL between `/d/` and `/edit`, pick the tab, map columns, then set duplicate handling, a label, and optionally a campaign to enroll into and categories to apply. **Nothing syncs automatically**: press **Sync now**, or save and sync immediately. Saved sources live in your Sync sources list to re-run, edit, or remove. Each sync dedupes on lowercased email using your chosen duplicate handling, so re-syncing a sheet with new rows is safe. diff --git a/docs/content/docs/guides/segments.mdx b/docs/content/docs/guides/segments.mdx index ba84cdbcc..b9685ff0d 100644 --- a/docs/content/docs/guides/segments.mdx +++ b/docs/content/docs/guides/segments.mdx @@ -42,19 +42,36 @@ Conditions decide membership, and two overrides sit on top of them: The segment header shows how many contacts are pinned in or out, and a **Pinned contacts** panel below it lists them; **Back to automatic** clears an override so the conditions decide again. A contact's own drawer has a **Segments** section that shows every segment, whether the contact is in it, and the same pin in, pin out and back-to-automatic controls. -Sequences can pin as well: the **Add to segment** and **Remove from segment** action steps apply the override to a contact as it moves through a campaign flow, so a positive reply can drop someone into a "warm" segment automatically. +Sequences can pin as well: the **Add to segment** and **Remove from segment** action steps apply the override to a contact as it moves through a campaign flow, so a positive reply can drop someone into a "warm" segment automatically. So can an import: the file import wizard's **Add to segments** picker on its Options step pins every contact in the file into the segments you choose, so a fresh list can land straight in the audience it was collected for. ## Using a segment - **Browse**: a segment page lists its current members with the same table, filters, detail drawer and bulk actions as the contacts page. -- **Add to campaign**: enrols every current member as a lead of the campaign you pick, from the segment page or with **From segment** on a campaign's Leads tab. Contacts already in that campaign are skipped, and a running campaign wakes up to schedule the new leads. This is a snapshot: contacts who join the segment later are not added until you run it again. +- **Add to campaign**: enrols every current member as a lead of the campaign you pick, from the segment page or with **From segment** on a campaign's Leads tab. Contacts already in that campaign are skipped, and a running campaign wakes up to schedule the new leads. This is a snapshot: contacts who join the segment later are not added until you run it again. To keep a campaign fed automatically, [link the segment](#linking-a-segment-to-a-campaign) instead. +- **Link to campaign**: attaches the segment to a campaign as a live audience, so contacts who join the segment later become leads on their own. See [below](#linking-a-segment-to-a-campaign). - **Filter**: the contact list's filter bar has a **Segment** pill, and the same scope carries into an export. - **Duplicate**: the segment menu copies a definition to start a variation from. - **Search and export**: the contact search and export accept `segment_ids`, so anything that takes a contact filter can be scoped to a segment. +## Linking a segment to a campaign + +Where **Add to campaign** copies today's members once, a linked segment is a live audience: the campaign keeps enrolling whoever the segment says belongs. Manage the links from the **Segments** button on a campaign's Leads tab; a campaign can link up to 20 segments. + +Linking enrols every current member as a lead immediately. From then on, any contact who enters the segment, a new contact, an import, a condition edit, a manual pin-in, or simply drifting into a date or engagement condition, is enrolled automatically; a background check picks up drift about every 2 minutes. + +Enrolment is additive. A contact who later falls out of the segment keeps their lead row and their progress; the link only ever adds. Detaching a segment stops future enrolment but leaves the leads it already added in place. + +The campaign reacts to growth by status: + +- **Active**: woken up to schedule the new leads straight away. +- **Finished**: restarted automatically, through the same launch checks as pressing play. +- **Paused** (or draft): the leads accumulate and send when the campaign runs. + +A segment linked to a campaign cannot be deleted; detach it from the campaign first. + ## Segments in the API -Everything above can be driven from the [API](/api/reference/contacts/#segments): list, create, update and delete segments, pin contacts in or out, look up a contact's segments, and enrol a segment into a campaign. Contact search and export take `segment_ids` to scope any contact query to a segment, so an external system (a signup form, a CRM sync) can keep a segment current and let campaigns pick it up from there. +Everything above can be driven from the [API](/api/reference/contacts/#segments): list, create, update and delete segments, pin contacts in or out, look up a contact's segments, enrol a segment into a campaign once, and [link segments to a campaign](/api/reference/campaigns/#list-linked-segments) as a live audience. Contact search and export take `segment_ids` to scope any contact query to a segment, so an external system (a signup form, a CRM sync) can keep a segment current and let campaigns pick it up from there. API calls address a segment by its ID. It is shown at the bottom of the segment page header (click it to copy), in the **Copy segment ID** entry of a segment's row menu on the Segments tab, and in every segment the API returns. Reads take the `READ_CONTACTS` key scope, writes `WRITE_CONTACTS`, and enrolling into a campaign `WRITE_CAMPAIGNS`. diff --git a/docs/content/docs/guides/workspace-export-import.mdx b/docs/content/docs/guides/workspace-export-import.mdx index bbe2a32ff..8805d21ec 100644 --- a/docs/content/docs/guides/workspace-export-import.mdx +++ b/docs/content/docs/guides/workspace-export-import.mdx @@ -15,7 +15,7 @@ The data is split into groups. Every export includes **Workspace**; the rest are |-------|----------| | Workspace | The organization, members, roles, teams, mailboxes, API keys, webhooks, and settings, including the website tracking site key. Always included | | Contacts | Contacts, categories, segments with their manual overrides, notes, activities, and the suppression list | -| Campaigns | Campaigns, sequences, senders, attachments, and per-campaign settings | +| Campaigns | Campaigns, sequences, senders, linked segments, attachments, and per-campaign settings | | CRM | Pipelines, deals, tasks, and meeting bookings | | Automations | Automations, connected integrations, and lead sync sources | | Assistant | Assistant sessions and messages, skills, MCP servers, and AI settings | diff --git a/internal/api/handler/segment.go b/internal/api/handler/segment.go index 2908e99ba..058cc6ead 100644 --- a/internal/api/handler/segment.go +++ b/internal/api/handler/segment.go @@ -207,6 +207,56 @@ func (h *Handler) AddSegmentToCampaign(c *gin.Context) { c.JSON(http.StatusOK, res) } +// campaignSegmentScope pulls the org and the campaign id from a campaigns route. +func campaignSegmentScope(c *gin.Context) (uuid.UUID, uuid.UUID, bool) { + orgID := middleware.GetOrganizationID(c) + if orgID == nil { + errx.Handle(c, errx.New(errx.BadRequest, "no organization selected")) + return uuid.Nil, uuid.Nil, false + } + id, err := uuid.Parse(c.Param("id")) + if err != nil { + errx.Handle(c, errx.New(errx.BadRequest, "invalid campaign id")) + return uuid.Nil, uuid.Nil, false + } + return *orgID, id, true +} + +// ListCampaignSegments lists the segments linked to a campaign. +func (h *Handler) ListCampaignSegments(c *gin.Context) { + orgID, campaignID, ok := campaignSegmentScope(c) + if !ok { + return + } + out, xerr := h.SegmentService.ListCampaignSegments(c.Request.Context(), orgID, campaignID) + if xerr != nil { + errx.Handle(c, xerr) + return + } + c.JSON(http.StatusOK, gin.H{"data": out}) +} + +// SetCampaignSegments replaces a campaign's linked segments. Members of the +// linked segments are enrolled as leads immediately and kept current. +func (h *Handler) SetCampaignSegments(c *gin.Context) { + orgID, campaignID, ok := campaignSegmentScope(c) + if !ok { + return + } + var in models.CampaignSegmentsWrite + if err := c.ShouldBindJSON(&in); err != nil { + errx.Handle(c, errx.ErrInvalid) + return + } + links, added, xerr := h.SegmentService.SetCampaignSegments(c.Request.Context(), orgID, campaignID, &in) + if xerr != nil { + errx.Handle(c, xerr) + return + } + h.auditOrg(c, models.AuditActionUpdate, models.AuditEntityCampaign, &campaignID, nil, map[string]string{"segments": itoa(len(links)), "added": itoa(added)}) + c.JSON(http.StatusOK, gin.H{"data": links, "added": added}) +} + // ListSegmentOverrides lists the contacts pinned into or out of a segment. func (h *Handler) ListSegmentOverrides(c *gin.Context) { orgID, id, ok := segmentScope(c) diff --git a/internal/api/routes.go b/internal/api/routes.go index f5ee38675..1877ecde4 100644 --- a/internal/api/routes.go +++ b/internal/api/routes.go @@ -486,6 +486,12 @@ func Run( campaigns.GET("/:id/senders", m.RequireOrganization(), m.RequireAccess(models.PermViewCampaigns, models.APIPermReadCampaigns), h.ListCampaignSenders) campaigns.PUT("/:id/senders", m.RequireOrganization(), m.RequireAccess(models.PermManageCampaigns, models.APIPermWriteCampaigns), h.ReplaceCampaignSenders) + // Linked segments: a live audience source; members are enrolled + // as leads automatically. PUT replaces the set, so retries are + // naturally safe. + campaigns.GET("/:id/segments", m.RequireOrganization(), m.RequireAccess(models.PermViewCampaigns, models.APIPermReadCampaigns), h.ListCampaignSegments) + campaigns.PUT("/:id/segments", m.RequireOrganization(), m.RequireAccess(models.PermManageCampaigns, models.APIPermWriteCampaigns), h.SetCampaignSegments) + // Campaign-scoped tracking-domain verification. campaigns.POST("/:id/tracking-domain/verify", m.RequireOrganization(), m.RequireAccess(models.PermManageCampaigns, models.APIPermWriteCampaigns), h.VerifyCampaignTrackingDomain) diff --git a/internal/app/contact/handler.go b/internal/app/contact/handler.go index 106d32218..263e6f9c6 100644 --- a/internal/app/contact/handler.go +++ b/internal/app/contact/handler.go @@ -67,6 +67,7 @@ func (s *contactService) Add(ctx context.Context, userID string, orgID uuid.UUID attached = append(attached, contacts[i].Campaigns...) } s.wakeCampaigns(ctx, orgID, attached) + s.syncSegmentCampaigns(ctx, orgID) return created, nil } @@ -135,6 +136,7 @@ func (s *contactService) BulkUpdate(ctx context.Context, userID string, orgID uu s.publishContactsReload(ctx, userID, "contacts:bulk_update") s.wakeCampaigns(ctx, orgID, data.AddCampaigns) + s.syncSegmentCampaigns(ctx, orgID) return updated, nil } @@ -146,6 +148,7 @@ func (s *contactService) Update(ctx context.Context, userID, contactID string, o s.publishContactsReload(ctx, userID, "contacts:update:"+contactID) s.wakeCampaigns(ctx, orgID, data.Campaigns) + s.syncSegmentCampaigns(ctx, orgID) return updated, nil } diff --git a/internal/app/contact/import.go b/internal/app/contact/import.go index 72eca850d..1daa2a5ac 100644 --- a/internal/app/contact/import.go +++ b/internal/app/contact/import.go @@ -222,6 +222,12 @@ func (s *contactService) ImportCommit( if xerr != nil { return nil, xerr } + // Segment targets are resolved up front: each must exist in the org. + // Membership is written after the rows exist, as an include override. + segmentIDs, xerr := s.parseSegmentIDs(ctx, orgID, opts.SegmentIDs) + if xerr != nil { + return nil, xerr + } rows, _, xerr := parseSpreadsheet(r, filename) if xerr != nil { @@ -381,6 +387,9 @@ func (s *contactService) ImportCommit( toInsert := make([]models.AddContact, 0, len(parsed)) toInsertLines := make([]int, 0, len(parsed)) toUpdate := make([]pendingRow, 0) + // Every contact the file touched (created, updated or skipped-but-linked) + // joins the import's target segments at the end. + touched := make([]uuid.UUID, 0, len(parsed)) // Existing contacts the file listed but did not change. They still have // to join the campaign / categories this import targets: "skip" means // "don't touch their fields", not "leave them out of the list". @@ -409,6 +418,7 @@ func (s *contactService) ImportCommit( toInsertLines = append(toInsertLines, p.line) case dedup == models.ContactImportDedupSkip: res.Skipped++ + touched = append(touched, ex.ID) skippedLinks = append(skippedLinks, linkTarget{ line: p.line, email: p.contact.Email, @@ -456,6 +466,9 @@ func (s *contactService) ImportCommit( continue } res.Imported += len(inserted) + for i := range inserted { + touched = append(touched, inserted[i].ID) + } } for _, p := range toUpdate { @@ -485,6 +498,7 @@ func (s *contactService) ImportCommit( continue } res.Updated++ + touched = append(touched, ex.ID) // Attach campaigns separately if the caller requested it. if len(p.contact.Campaigns) > 0 { @@ -522,6 +536,16 @@ func (s *contactService) ImportCommit( } } + // Segment membership last, once every touched row exists. A failed write + // is a note, not a failed import: the contacts themselves are in. + if len(touched) > 0 { + for _, segID := range segmentIDs { + if _, xerr := s.segmentLinker.SetMembers(ctx, orgID, segID, touched, models.SegmentMemberInclude); xerr != nil { + warn(0, "", nil, "imported contacts could not be added to a segment: "+xerr.Message) + } + } + } + res.EndedAt = time.Now().UTC() s.updateListQuality(ctx, orgID, quality) @@ -530,10 +554,42 @@ func (s *contactService) ImportCommit( s.publishContactsReload(ctx, userID, "contacts:import") // Covers the Google Sheets sync too: it commits through this path. s.wakeCampaigns(ctx, orgID, globalCampaignIDs) + s.syncSegmentCampaigns(ctx, orgID) } return res, nil } +// parseSegmentIDs validates the import's target segments: well-formed ids +// that exist in the org, deduplicated. +func (s *contactService) parseSegmentIDs(ctx context.Context, orgID uuid.UUID, raw []string) ([]uuid.UUID, *errx.Error) { + if len(raw) == 0 { + return nil, nil + } + if s.segmentLinker == nil { + return nil, errx.New(errx.BadRequest, "segments are not available") + } + out := make([]uuid.UUID, 0, len(raw)) + seen := map[uuid.UUID]bool{} + for _, r := range raw { + id, err := uuid.Parse(strings.TrimSpace(r)) + if err != nil { + return nil, errx.New(errx.BadRequest, "invalid segment id") + } + if seen[id] { + continue + } + seen[id] = true + if _, xerr := s.segmentLinker.Get(ctx, orgID, id); xerr != nil { + if xerr.Code == errx.NotFound { + return nil, errx.New(errx.BadRequest, "a selected segment does not exist") + } + return nil, xerr + } + out = append(out, id) + } + return out, nil +} + // linkTarget is an existing contact that must join the import's campaigns and // categories even though its own fields were left alone. type linkTarget struct { diff --git a/internal/app/contact/service.go b/internal/app/contact/service.go index 55ba9df10..6cc28ba8b 100644 --- a/internal/app/contact/service.go +++ b/internal/app/contact/service.go @@ -91,12 +91,32 @@ type CampaignWaker interface { WakeCampaigns(ctx context.Context, orgID uuid.UUID, campaignIDs []string) } +// SegmentLinker writes segment include overrides for imported contacts. +// Satisfied structurally by repository.SegmentRepository. +type SegmentLinker interface { + Get(ctx context.Context, orgID, id uuid.UUID) (*models.Segment, *errx.Error) + SetMembers(ctx context.Context, orgID, segmentID uuid.UUID, contactIDs []uuid.UUID, mode models.SegmentMemberMode) (int, *errx.Error) +} + +// SegmentCampaignSyncer re-enrols the org's segment-linked campaigns after +// contacts changed. Satisfied structurally by segment.Service. +type SegmentCampaignSyncer interface { + SyncOrgLinkedCampaigns(ctx context.Context, orgID uuid.UUID) +} + +// SegmentAware is the optional capability the caller uses to attach both. +type SegmentAware interface { + WireSegments(linker SegmentLinker, syncer SegmentCampaignSyncer) +} + type contactService struct { contactRepository repository.ContactRepository subRepo repository.SubscriptionRepository planRepo repository.PlanRepository streamingPublisher *pubsub.StreamingPublisher campaignWaker CampaignWaker + segmentLinker SegmentLinker + segmentSyncer SegmentCampaignSyncer previewer scheduler.ContactSendPreviewer // orgRisk files import-quality findings on the workspace's posture. // Optional/nil-safe: without it a bad import is reported but not fused. @@ -155,6 +175,22 @@ func (s *contactService) publishContactsReload(ctx context.Context, userID strin func (s *contactService) SetCampaignWaker(w CampaignWaker) { s.campaignWaker = w } +// WireSegments attaches the segment override writer and linked-campaign syncer. +func (s *contactService) WireSegments(linker SegmentLinker, syncer SegmentCampaignSyncer) { + s.segmentLinker = linker + s.segmentSyncer = syncer +} + +// syncSegmentCampaigns follows every contact-mutating path: a changed contact +// may now match a segment linked to a campaign. Best effort like wakeCampaigns; +// the periodic sweep is the backstop. +func (s *contactService) syncSegmentCampaigns(ctx context.Context, orgID uuid.UUID) { + if s.segmentSyncer == nil { + return + } + s.segmentSyncer.SyncOrgLinkedCampaigns(ctx, orgID) +} + // wakeCampaigns is the single place every lead-attaching path funnels through // (add, update, bulk edit, CSV/XLSX import and the Google Sheets sync, which // runs through ImportCommit). Best effort: the lead is already stored, and the diff --git a/internal/app/orgtransfer/spec.go b/internal/app/orgtransfer/spec.go index daad34c7a..32d90a714 100644 --- a/internal/app/orgtransfer/spec.go +++ b/internal/app/orgtransfer/spec.go @@ -346,6 +346,12 @@ var Tables = []Table{ Name: "campaign_leads", Group: models.OrgDataGroupCampaigns, Scope: `campaign_id IN ` + orgCampaigns, }, + { + // Segments travel in the contacts group, which campaigns already + // require, so both ends of the link exist by the time this applies. + Name: "campaign_segments", Group: models.OrgDataGroupCampaigns, + Scope: `campaign_id IN ` + orgCampaigns, + }, { Name: "campaign_ab_assignments", Group: models.OrgDataGroupCampaigns, Scope: `campaign_id IN ` + orgCampaigns, diff --git a/internal/app/segment/service.go b/internal/app/segment/service.go index 17275c722..5d6e69469 100644 --- a/internal/app/segment/service.go +++ b/internal/app/segment/service.go @@ -1,13 +1,17 @@ // Package segment manages saved contact audiences; membership is computed at -// read time, so nothing here schedules work. +// read time. The one exception is campaign links: a segment attached to a +// campaign enrols its members as leads, kept current by targeted syncs on +// every membership-changing write plus a periodic sweep. package segment import ( "context" "fmt" "strings" + "time" "github.com/google/uuid" + "github.com/rs/zerolog/log" "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/models" "github.com/warmbly/warmbly/internal/repository" @@ -18,6 +22,12 @@ type CampaignWaker interface { WakeCampaigns(ctx context.Context, orgID uuid.UUID, campaignIDs []string) } +// CampaignStarter restarts a completed campaign whose linked segments grew. +// Satisfied structurally by campaign.CampaignService. +type CampaignStarter interface { + StartCampaign(ctx context.Context, orgID uuid.UUID, campaignID string, opts models.StartCampaignOptions) *errx.Error +} + type Service interface { List(ctx context.Context, orgID uuid.UUID) ([]models.Segment, *errx.Error) Get(ctx context.Context, orgID, id uuid.UUID) (*models.Segment, *errx.Error) @@ -32,7 +42,20 @@ type Service interface { Fields(ctx context.Context, orgID uuid.UUID) ([]models.SegmentFieldSpec, *errx.Error) SegmentsForContact(ctx context.Context, orgID, contactID uuid.UUID) ([]models.ContactSegment, *errx.Error) Overrides(ctx context.Context, orgID, id uuid.UUID) ([]models.SegmentOverride, *errx.Error) + // ListCampaignSegments lists the segments linked to a campaign. + ListCampaignSegments(ctx context.Context, orgID, campaignID uuid.UUID) ([]models.CampaignSegmentLink, *errx.Error) + // SetCampaignSegments replaces a campaign's linked segments and enrols + // their current members immediately. Returns the links and how many + // leads were new. + SetCampaignSegments(ctx context.Context, orgID, campaignID uuid.UUID, in *models.CampaignSegmentsWrite) ([]models.CampaignSegmentLink, int, *errx.Error) + // SyncOrgLinkedCampaigns re-enrols every linked campaign of the org after + // contacts changed. Best effort: the sweep is the backstop. + SyncOrgLinkedCampaigns(ctx context.Context, orgID uuid.UUID) + // StartCampaignSegmentSync sweeps every linked campaign on an interval so + // membership drift (dates, engagement, nested segments) still enrols. + StartCampaignSegmentSync(ctx context.Context, interval time.Duration) SetCampaignWaker(w CampaignWaker) + SetCampaignStarter(st CampaignStarter) } // CustomFieldLister is the slice of the contact repository Fields needs. @@ -41,16 +64,18 @@ type CustomFieldLister interface { } type service struct { - repo repository.SegmentRepository - fields CustomFieldLister - waker CampaignWaker + repo repository.SegmentRepository + fields CustomFieldLister + waker CampaignWaker + starter CampaignStarter } func NewService(repo repository.SegmentRepository, fields CustomFieldLister) Service { return &service{repo: repo, fields: fields} } -func (s *service) SetCampaignWaker(w CampaignWaker) { s.waker = w } +func (s *service) SetCampaignWaker(w CampaignWaker) { s.waker = w } +func (s *service) SetCampaignStarter(st CampaignStarter) { s.starter = st } func (s *service) List(ctx context.Context, orgID uuid.UUID) ([]models.Segment, *errx.Error) { return s.repo.List(ctx, orgID) @@ -166,7 +191,16 @@ func (s *service) Update(ctx context.Context, orgID, id uuid.UUID, in *models.Se return nil, xerr } } - return s.repo.Update(ctx, orgID, seg) + out, xerr := s.repo.Update(ctx, orgID, seg) + if xerr != nil { + return nil, xerr + } + // A definition change can admit new members; enrol them into linked + // campaigns right away instead of waiting for the sweep. + if in.Conditions != nil || in.Match != nil { + s.syncLinkedCampaignsForSegments(ctx, orgID, []uuid.UUID{id}) + } + return out, nil } func (s *service) Delete(ctx context.Context, orgID, id uuid.UUID) *errx.Error { @@ -177,6 +211,13 @@ func (s *service) Delete(ctx context.Context, orgID, id uuid.UUID) *errx.Error { if len(names) > 0 { return errx.New(errx.Conflict, "this segment is used by: "+strings.Join(names, ", ")+". Remove it from those segments first") } + campaigns, xerr := s.repo.CampaignsUsingSegment(ctx, orgID, id) + if xerr != nil { + return xerr + } + if len(campaigns) > 0 { + return errx.New(errx.Conflict, "this segment is linked to: "+strings.Join(campaigns, ", ")+". Detach it from those campaigns first") + } return s.repo.Delete(ctx, orgID, id) } @@ -235,7 +276,15 @@ func (s *service) SetMembers(ctx context.Context, orgID, id uuid.UUID, in *model if _, xerr := s.repo.Get(ctx, orgID, id); xerr != nil { return 0, xerr } - return s.repo.SetMembers(ctx, orgID, id, ids, in.Mode) + n, xerr := s.repo.SetMembers(ctx, orgID, id, ids, in.Mode) + if xerr != nil { + return 0, xerr + } + // A pin-in (or a cleared exclude) can admit new members. + if in.Mode != models.SegmentMemberExclude { + s.syncLinkedCampaignsForSegments(ctx, orgID, []uuid.UUID{id}) + } + return n, nil } func (s *service) MemberModes(ctx context.Context, orgID, id uuid.UUID, contactIDs []string) (map[uuid.UUID]models.SegmentMemberMode, *errx.Error) { @@ -264,6 +313,136 @@ func (s *service) AddToCampaign(ctx context.Context, orgID uuid.UUID, actor stri return res, nil } +func (s *service) ListCampaignSegments(ctx context.Context, orgID, campaignID uuid.UUID) ([]models.CampaignSegmentLink, *errx.Error) { + return s.repo.ListForCampaign(ctx, orgID, campaignID) +} + +func (s *service) SetCampaignSegments(ctx context.Context, orgID, campaignID uuid.UUID, in *models.CampaignSegmentsWrite) ([]models.CampaignSegmentLink, int, *errx.Error) { + seen := map[uuid.UUID]bool{} + ids := make([]uuid.UUID, 0, len(in.SegmentIDs)) + for _, raw := range in.SegmentIDs { + id, err := uuid.Parse(raw) + if err != nil { + return nil, 0, errx.New(errx.BadRequest, "invalid segment id") + } + if seen[id] { + continue + } + seen[id] = true + ids = append(ids, id) + } + if len(ids) > models.CampaignSegmentsMax { + return nil, 0, errx.New(errx.BadRequest, fmt.Sprintf("a campaign can link at most %d segments", models.CampaignSegmentsMax)) + } + if xerr := s.repo.SetForCampaign(ctx, orgID, campaignID, ids); xerr != nil { + return nil, 0, xerr + } + added := 0 + if len(ids) > 0 { + links, xerr := s.repo.LinkedCampaignsForSegments(ctx, orgID, ids) + if xerr != nil { + return nil, 0, xerr + } + for _, lc := range links { + if lc.CampaignID == campaignID { + added = s.syncLinkedCampaign(ctx, lc) + } + } + } + out, xerr := s.repo.ListForCampaign(ctx, orgID, campaignID) + if xerr != nil { + return nil, 0, xerr + } + return out, added, nil +} + +// syncLinkedCampaign enrols missing leads for one linked campaign, waking an +// active chain and restarting a completed one when anything was added. +func (s *service) syncLinkedCampaign(ctx context.Context, lc models.LinkedCampaign) int { + added, xerr := s.repo.SyncCampaignSegments(ctx, lc.OrganizationID, lc.CampaignID) + if xerr != nil { + log.Warn().Str("campaign_id", lc.CampaignID.String()).Str("error", xerr.Message).Msg("segment sync: enrol failed") + return 0 + } + if added == 0 { + return 0 + } + switch lc.Status { + case "active": + if s.waker != nil { + s.waker.WakeCampaigns(ctx, lc.OrganizationID, []string{lc.CampaignID.String()}) + } + case "completed": + // Completed only means the campaign ran out of leads; new segment + // members are exactly the reason to pick it back up. StartCampaign + // re-runs every launch check, so a campaign past its end date, over + // plan limits, or with a risky list stays closed. + if s.starter != nil { + if xerr := s.starter.StartCampaign(ctx, lc.OrganizationID, lc.CampaignID.String(), models.StartCampaignOptions{Automatic: true}); xerr != nil { + log.Info().Str("campaign_id", lc.CampaignID.String()).Str("reason", xerr.Message).Msg("segment sync: completed campaign not restarted") + } + } + } + return added +} + +// syncLinkedCampaignsForSegments re-enrols the campaigns linked to any of the +// given segments. Nested references (a linked segment built on this one) are +// not chased here; the periodic sweep covers them. +func (s *service) syncLinkedCampaignsForSegments(ctx context.Context, orgID uuid.UUID, segmentIDs []uuid.UUID) { + links, xerr := s.repo.LinkedCampaignsForSegments(ctx, orgID, segmentIDs) + if xerr != nil { + return + } + for _, lc := range links { + s.syncLinkedCampaign(ctx, lc) + } +} + +func (s *service) SyncOrgLinkedCampaigns(ctx context.Context, orgID uuid.UUID) { + links, xerr := s.repo.LinkedCampaigns(ctx, &orgID) + if xerr != nil { + return + } + for _, lc := range links { + s.syncLinkedCampaign(ctx, lc) + } +} + +func (s *service) StartCampaignSegmentSync(ctx context.Context, interval time.Duration) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + + // Sweep once on boot so links catch up promptly after a restart. + s.sweepLinkedCampaigns(ctx) + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + s.sweepLinkedCampaigns(ctx) + } + } +} + +func (s *service) sweepLinkedCampaigns(ctx context.Context) { + rctx, cancel := context.WithTimeout(ctx, 2*time.Minute) + defer cancel() + links, xerr := s.repo.LinkedCampaigns(rctx, nil) + if xerr != nil { + log.Warn().Str("error", xerr.Message).Msg("segment sync: sweep scan failed") + return + } + total := 0 + for _, lc := range links { + total += s.syncLinkedCampaign(rctx, lc) + } + if total > 0 { + log.Info().Int("added", total).Msg("segment sync: sweep enrolled new leads") + } +} + func (s *service) Fields(ctx context.Context, orgID uuid.UUID) ([]models.SegmentFieldSpec, *errx.Error) { out := make([]models.SegmentFieldSpec, 0, len(models.SegmentFieldCatalog)+16) out = append(out, models.SegmentFieldCatalog...) diff --git a/internal/infrastructure/db/migrations/000114_campaign_segments.down.sql b/internal/infrastructure/db/migrations/000114_campaign_segments.down.sql new file mode 100644 index 000000000..568be908e --- /dev/null +++ b/internal/infrastructure/db/migrations/000114_campaign_segments.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS campaign_segments; diff --git a/internal/infrastructure/db/migrations/000114_campaign_segments.up.sql b/internal/infrastructure/db/migrations/000114_campaign_segments.up.sql new file mode 100644 index 000000000..f84573511 --- /dev/null +++ b/internal/infrastructure/db/migrations/000114_campaign_segments.up.sql @@ -0,0 +1,11 @@ +-- Live segment -> campaign links. Contacts matching an attached segment are +-- enrolled as campaign leads automatically; enrolment is additive, so a +-- contact falling out of the segment keeps their lead row. +CREATE TABLE campaign_segments ( + campaign_id uuid NOT NULL REFERENCES campaigns (id) ON DELETE CASCADE, + segment_id uuid NOT NULL REFERENCES segments (id) ON DELETE CASCADE, + created_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (campaign_id, segment_id) +); + +CREATE INDEX idx_campaign_segments_segment ON campaign_segments (segment_id); diff --git a/internal/models/contact_import.go b/internal/models/contact_import.go index 92513f8ad..10103aad7 100644 --- a/internal/models/contact_import.go +++ b/internal/models/contact_import.go @@ -99,13 +99,16 @@ type ContactImportPreview struct { // ContactImportCommit is the full configuration for committing an // import: how to map columns, how to treat collisions, what categories // to assign, what the default subscription state is, and which -// campaign(s) the imported contacts should join. +// campaign(s) and segment(s) the imported contacts should join. type ContactImportCommit struct { Mapping []ContactImportColumnMapping `json:"mapping"` Dedup ContactImportDedupStrategy `json:"dedup"` HasHeader bool `json:"has_header"` CategoryIDs []string `json:"category_ids,omitempty"` CampaignIDs []string `json:"campaign_ids,omitempty"` + // SegmentIDs pins every imported row into these segments as a manual + // include override, the same write the "Add to segment" bulk action does. + SegmentIDs []string `json:"segment_ids,omitempty"` // SubscribedDefault is what new contacts inherit when no // subscribed column was mapped. Defaults to true server-side. diff --git a/internal/models/segment.go b/internal/models/segment.go index 363a88ae1..483442f10 100644 --- a/internal/models/segment.go +++ b/internal/models/segment.go @@ -235,6 +235,32 @@ type SegmentAddToCampaignResult struct { Members int `json:"members"` } +// CampaignSegmentsMax bounds how many segments one campaign can link. +const CampaignSegmentsMax = 20 + +// CampaignSegmentsWrite replaces a campaign's linked segments. +type CampaignSegmentsWrite struct { + SegmentIDs []string `json:"segment_ids"` +} + +// CampaignSegmentLink is one segment linked to a campaign, for the Leads tab. +type CampaignSegmentLink struct { + SegmentID uuid.UUID `json:"segment_id"` + Name string `json:"name"` + Color string `json:"color"` + Description string `json:"description"` + ContactCount int `json:"contact_count"` + LinkedAt time.Time `json:"linked_at"` +} + +// LinkedCampaign is one campaign that has segments attached, for the sync +// sweep and the targeted per-segment syncs. +type LinkedCampaign struct { + CampaignID uuid.UUID + OrganizationID uuid.UUID + Status string +} + // ContactSegment is one segment a contact belongs to, with its override. type ContactSegment struct { ID uuid.UUID `json:"id"` diff --git a/internal/repository/pg_segment.go b/internal/repository/pg_segment.go index 5abbb7331..22f1a11a4 100644 --- a/internal/repository/pg_segment.go +++ b/internal/repository/pg_segment.go @@ -30,6 +30,20 @@ type SegmentRepository interface { MemberModes(ctx context.Context, segmentID uuid.UUID, contactIDs []uuid.UUID) (map[uuid.UUID]models.SegmentMemberMode, *errx.Error) // AddToCampaign enrols every current member of the segment as a lead. AddToCampaign(ctx context.Context, orgID uuid.UUID, actor string, segmentID, campaignID uuid.UUID) (*models.SegmentAddToCampaignResult, *errx.Error) + // ListForCampaign lists the segments linked to a campaign. + ListForCampaign(ctx context.Context, orgID, campaignID uuid.UUID) ([]models.CampaignSegmentLink, *errx.Error) + // SetForCampaign replaces the campaign's linked segments. + SetForCampaign(ctx context.Context, orgID, campaignID uuid.UUID, segmentIDs []uuid.UUID) *errx.Error + // SyncCampaignSegments enrols every current member of the campaign's + // linked segments that is not yet a lead; returns how many were added. + SyncCampaignSegments(ctx context.Context, orgID, campaignID uuid.UUID) (int, *errx.Error) + // LinkedCampaigns lists campaigns with linked segments; nil orgID sweeps + // the whole instance. + LinkedCampaigns(ctx context.Context, orgID *uuid.UUID) ([]models.LinkedCampaign, *errx.Error) + // LinkedCampaignsForSegments lists the campaigns any of the segments link to. + LinkedCampaignsForSegments(ctx context.Context, orgID uuid.UUID, segmentIDs []uuid.UUID) ([]models.LinkedCampaign, *errx.Error) + // CampaignsUsingSegment names the campaigns a segment is linked to. + CampaignsUsingSegment(ctx context.Context, orgID, segmentID uuid.UUID) ([]string, *errx.Error) // SegmentsForContact evaluates every segment of the org for one contact. SegmentsForContact(ctx context.Context, orgID, contactID uuid.UUID) ([]models.ContactSegment, *errx.Error) // ListOverrides lists the manually included and excluded contacts. @@ -285,25 +299,9 @@ func (r *segmentRepository) AddToCampaign(ctx context.Context, orgID uuid.UUID, return nil, errx.InternalError() } - // The campaign is bound after the compiled clause so the count query - // above carries no unused parameter. - args = append(args, campaignID) - insertQ := fmt.Sprintf(`INSERT INTO campaign_leads (contact_id, campaign_id) - SELECT c.id, $%d::uuid FROM contacts c WHERE c.organization_id = $1 AND (%s) - ON CONFLICT DO NOTHING - RETURNING contact_id, campaign_id`, len(args), clause) - rows, err := tx.Query(ctx, insertQ, args...) + links, err := insertSegmentLeads(ctx, tx, orgID, actorID(actor), clause, args, campaignID) if err != nil { - db.CaptureError(err, insertQ, args, "query") - return nil, errx.InternalError() - } - links, err := collectLinkPairs(rows) - if err != nil { - db.CaptureError(err, insertQ, args, "returning") - return nil, errx.InternalError() - } - if err := logCampaignLinks(ctx, tx, orgID, actorID(actor), models.ActivityCampaignAdded, links); err != nil { - db.CaptureError(err, "", nil, "campaign_added activity") + db.CaptureError(err, "segment enrol", nil, "query") return nil, errx.InternalError() } if err := tx.Commit(ctx); err != nil { @@ -313,6 +311,253 @@ func (r *segmentRepository) AddToCampaign(ctx context.Context, orgID uuid.UUID, return &models.SegmentAddToCampaignResult{CampaignID: campaignID, Added: len(links), Members: members}, nil } +// insertSegmentLeads enrols every contact matching the precompiled segment +// clause as a lead, logging a campaign_added activity for each row that was +// actually new. The campaign is bound after the clause's own parameters. +func insertSegmentLeads(ctx context.Context, tx pgx.Tx, orgID uuid.UUID, actor *uuid.UUID, clause string, args []any, campaignID uuid.UUID) ([]contactLink, error) { + args = append(args, campaignID) + insertQ := fmt.Sprintf(`INSERT INTO campaign_leads (contact_id, campaign_id) + SELECT c.id, $%d::uuid FROM contacts c WHERE c.organization_id = $1 AND (%s) + ON CONFLICT DO NOTHING + RETURNING contact_id, campaign_id`, len(args), clause) + rows, err := tx.Query(ctx, insertQ, args...) + if err != nil { + return nil, err + } + links, err := collectLinkPairs(rows) + if err != nil { + return nil, err + } + if err := logCampaignLinks(ctx, tx, orgID, actor, models.ActivityCampaignAdded, links); err != nil { + return nil, err + } + return links, nil +} + +func (r *segmentRepository) ListForCampaign(ctx context.Context, orgID, campaignID uuid.UUID) ([]models.CampaignSegmentLink, *errx.Error) { + var exists bool + if err := r.DB.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM campaigns WHERE id = $1 AND organization_id = $2)`, campaignID, orgID).Scan(&exists); err != nil { + db.CaptureError(err, "campaign exists", nil, "queryrow") + return nil, errx.InternalError() + } + if !exists { + return nil, errx.New(errx.NotFound, "campaign not found") + } + rows, err := r.DB.Query(ctx, ` + SELECT s.id, s.name, s.color, s.description, s.match, s.conditions, cs.created_at + FROM campaign_segments cs + JOIN segments s ON s.id = cs.segment_id + WHERE cs.campaign_id = $1 + ORDER BY lower(s.name) ASC`, campaignID) + if err != nil { + db.CaptureError(err, "campaign segments list", nil, "query") + return nil, errx.InternalError() + } + defer rows.Close() + out := []models.CampaignSegmentLink{} + // Held alongside so the live counts below evaluate the same definitions. + var matches []models.SegmentMatch + var conds [][]models.SegmentCondition + for rows.Next() { + var l models.CampaignSegmentLink + var match string + var raw []byte + if err := rows.Scan(&l.SegmentID, &l.Name, &l.Color, &l.Description, &match, &raw, &l.LinkedAt); err != nil { + db.CaptureError(err, "", nil, "scan") + return nil, errx.InternalError() + } + cs := []models.SegmentCondition{} + if len(raw) > 0 { + if err := json.Unmarshal(raw, &cs); err != nil { + db.CaptureError(err, "", nil, "scan") + return nil, errx.InternalError() + } + } + out = append(out, l) + matches = append(matches, models.SegmentMatch(match)) + conds = append(conds, cs) + } + for i := range out { + n, xerr := r.Count(ctx, orgID, &out[i].SegmentID, matches[i], conds[i]) + if xerr != nil { + return nil, xerr + } + out[i].ContactCount = n + } + return out, nil +} + +func (r *segmentRepository) SetForCampaign(ctx context.Context, orgID, campaignID uuid.UUID, segmentIDs []uuid.UUID) *errx.Error { + // A nil slice would reach Postgres as ANY(NULL) and skip the delete. + if segmentIDs == nil { + segmentIDs = []uuid.UUID{} + } + tx, err := r.DB.Begin(ctx) + if err != nil { + db.CaptureError(err, "", nil, "begin") + return errx.InternalError() + } + defer tx.Rollback(ctx) + + var exists bool + if err := tx.QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM campaigns WHERE id = $1 AND organization_id = $2)`, campaignID, orgID).Scan(&exists); err != nil { + db.CaptureError(err, "campaign exists", nil, "queryrow") + return errx.InternalError() + } + if !exists { + return errx.New(errx.NotFound, "campaign not found") + } + if len(segmentIDs) > 0 { + var n int + if err := tx.QueryRow(ctx, `SELECT COUNT(*) FROM segments WHERE organization_id = $1 AND id = ANY($2::uuid[])`, orgID, segmentIDs).Scan(&n); err != nil { + db.CaptureError(err, "segments verify", nil, "queryrow") + return errx.InternalError() + } + if n != len(segmentIDs) { + return errx.New(errx.BadRequest, "a linked segment does not exist") + } + } + if _, err := tx.Exec(ctx, `DELETE FROM campaign_segments WHERE campaign_id = $1 AND NOT (segment_id = ANY($2::uuid[]))`, campaignID, segmentIDs); err != nil { + db.CaptureError(err, "campaign segments delete", nil, "exec") + return errx.InternalError() + } + if len(segmentIDs) > 0 { + if _, err := tx.Exec(ctx, `INSERT INTO campaign_segments (campaign_id, segment_id) SELECT $1, unnest($2::uuid[]) ON CONFLICT DO NOTHING`, campaignID, segmentIDs); err != nil { + db.CaptureError(err, "campaign segments insert", nil, "exec") + return errx.InternalError() + } + } + if err := tx.Commit(ctx); err != nil { + db.CaptureError(err, "", nil, "commit") + return errx.InternalError() + } + return nil +} + +func (r *segmentRepository) SyncCampaignSegments(ctx context.Context, orgID, campaignID uuid.UUID) (int, *errx.Error) { + tx, err := r.DB.Begin(ctx) + if err != nil { + db.CaptureError(err, "", nil, "begin") + return 0, errx.InternalError() + } + defer tx.Rollback(ctx) + + rows, err := tx.Query(ctx, ` + SELECT cs.segment_id FROM campaign_segments cs + JOIN campaigns cp ON cp.id = cs.campaign_id + WHERE cp.id = $1 AND cp.organization_id = $2`, campaignID, orgID) + if err != nil { + db.CaptureError(err, "campaign segments sync", nil, "query") + return 0, errx.InternalError() + } + var segmentIDs []uuid.UUID + for rows.Next() { + var id uuid.UUID + if err := rows.Scan(&id); err != nil { + rows.Close() + db.CaptureError(err, "", nil, "scan") + return 0, errx.InternalError() + } + segmentIDs = append(segmentIDs, id) + } + rows.Close() + + total := 0 + for _, segID := range segmentIDs { + args := []any{orgID} + clause, args, cerr := compileSavedSegment(ctx, tx, orgID, segID, args) + if cerr != nil { + db.CaptureError(cerr, "segment compile", nil, "query") + return 0, errx.InternalError() + } + // FALSE means the segment vanished mid-pass; the link row follows it. + if clause == "FALSE" { + continue + } + links, lerr := insertSegmentLeads(ctx, tx, orgID, nil, clause, args, campaignID) + if lerr != nil { + db.CaptureError(lerr, "segment enrol", nil, "query") + return 0, errx.InternalError() + } + total += len(links) + } + if err := tx.Commit(ctx); err != nil { + db.CaptureError(err, "", nil, "commit") + return 0, errx.InternalError() + } + return total, nil +} + +const linkedCampaignsQuery = ` + SELECT DISTINCT cs.campaign_id, cp.organization_id, cp.status + FROM campaign_segments cs + JOIN campaigns cp ON cp.id = cs.campaign_id + WHERE cp.organization_id IS NOT NULL` + +func scanLinkedCampaigns(rows pgx.Rows) ([]models.LinkedCampaign, *errx.Error) { + defer rows.Close() + out := []models.LinkedCampaign{} + for rows.Next() { + var l models.LinkedCampaign + if err := rows.Scan(&l.CampaignID, &l.OrganizationID, &l.Status); err != nil { + db.CaptureError(err, "", nil, "scan") + return nil, errx.InternalError() + } + out = append(out, l) + } + return out, nil +} + +func (r *segmentRepository) LinkedCampaigns(ctx context.Context, orgID *uuid.UUID) ([]models.LinkedCampaign, *errx.Error) { + q := linkedCampaignsQuery + args := []any{} + if orgID != nil { + q += ` AND cp.organization_id = $1` + args = append(args, *orgID) + } + rows, err := r.DB.Query(ctx, q, args...) + if err != nil { + db.CaptureError(err, "linked campaigns", nil, "query") + return nil, errx.InternalError() + } + return scanLinkedCampaigns(rows) +} + +func (r *segmentRepository) LinkedCampaignsForSegments(ctx context.Context, orgID uuid.UUID, segmentIDs []uuid.UUID) ([]models.LinkedCampaign, *errx.Error) { + if len(segmentIDs) == 0 { + return []models.LinkedCampaign{}, nil + } + rows, err := r.DB.Query(ctx, linkedCampaignsQuery+` AND cp.organization_id = $1 AND cs.segment_id = ANY($2::uuid[])`, orgID, segmentIDs) + if err != nil { + db.CaptureError(err, "linked campaigns for segments", nil, "query") + return nil, errx.InternalError() + } + return scanLinkedCampaigns(rows) +} + +func (r *segmentRepository) CampaignsUsingSegment(ctx context.Context, orgID, segmentID uuid.UUID) ([]string, *errx.Error) { + rows, err := r.DB.Query(ctx, ` + SELECT cp.name FROM campaign_segments cs + JOIN campaigns cp ON cp.id = cs.campaign_id + WHERE cp.organization_id = $1 AND cs.segment_id = $2 + ORDER BY lower(cp.name)`, orgID, segmentID) + if err != nil { + db.CaptureError(err, "campaigns using segment", nil, "query") + return nil, errx.InternalError() + } + defer rows.Close() + var names []string + for rows.Next() { + var n string + if err := rows.Scan(&n); err != nil { + db.CaptureError(err, "", nil, "scan") + return nil, errx.InternalError() + } + names = append(names, n) + } + return names, nil +} + func (r *segmentRepository) SegmentsForContact(ctx context.Context, orgID, contactID uuid.UUID) ([]models.ContactSegment, *errx.Error) { rows, err := r.DB.Query(ctx, `SELECT `+segmentColumns+` FROM segments s WHERE s.organization_id = $1 ORDER BY lower(s.name) ASC`, orgID) if err != nil { diff --git a/internal/repository/segment_live_test.go b/internal/repository/segment_live_test.go index 158928006..0d0808b84 100644 --- a/internal/repository/segment_live_test.go +++ b/internal/repository/segment_live_test.go @@ -241,3 +241,75 @@ func TestLiveSegmentOverridesNestingAndCampaign(t *testing.T) { t.Fatalf("acme after nested delete: %+v", got) } } + +// Issue #277: segments linked to a campaign are a live audience source. These +// prove the link CRUD, the incremental enrolment sync, the sweep scan and the +// delete guard against the real schema. +func TestLiveSegmentCampaignLinks(t *testing.T) { + f, repo := newSegmentFixture(t) + ctx := context.Background() + + acme, xerr := repo.Create(ctx, f.org, &f.owner, &models.Segment{ + Name: "Acme live", Color: "#0284c7", Match: models.SegmentMatchAll, + Conditions: []models.SegmentCondition{{Field: "company", Operator: "equals", Value: "acme"}}, + }) + if xerr != nil { + t.Fatalf("create: %v", xerr) + } + + // Linking replaces the set; a segment the org does not own is refused. + if xerr := repo.SetForCampaign(ctx, f.org, f.other, []uuid.UUID{acme.ID}); xerr != nil { + t.Fatalf("set: %v", xerr) + } + if xerr := repo.SetForCampaign(ctx, f.org, f.other, []uuid.UUID{uuid.New()}); xerr == nil { + t.Fatalf("unknown segment accepted") + } + links, xerr := repo.ListForCampaign(ctx, f.org, f.other) + if xerr != nil || len(links) != 1 || links[0].SegmentID != acme.ID || links[0].ContactCount != 2 { + t.Fatalf("links = %+v, %v", links, xerr) + } + + // Sync enrols current members once; a second pass adds nothing. + added, xerr := repo.SyncCampaignSegments(ctx, f.org, f.other) + if xerr != nil || added != 2 { + t.Fatalf("sync = %d, %v", added, xerr) + } + if added, _ = repo.SyncCampaignSegments(ctx, f.org, f.other); added != 0 { + t.Fatalf("second sync = %d, want 0", added) + } + + // A contact pinned into the segment is picked up by the next sync. + if _, xerr := repo.SetMembers(ctx, f.org, acme.ID, []uuid.UUID{f.bob}, models.SegmentMemberInclude); xerr != nil { + t.Fatalf("include: %v", xerr) + } + if added, _ = repo.SyncCampaignSegments(ctx, f.org, f.other); added != 1 { + t.Fatalf("post-include sync = %d, want 1", added) + } + + // The sweep and the targeted lookup both see the linked campaign, and the + // delete guard reports it by name. + linked, xerr := repo.LinkedCampaigns(ctx, &f.org) + if xerr != nil || len(linked) != 1 || linked[0].CampaignID != f.other || linked[0].OrganizationID != f.org { + t.Fatalf("linked = %+v, %v", linked, xerr) + } + byseg, xerr := repo.LinkedCampaignsForSegments(ctx, f.org, []uuid.UUID{acme.ID}) + if xerr != nil || len(byseg) != 1 || byseg[0].CampaignID != f.other { + t.Fatalf("by segment = %+v, %v", byseg, xerr) + } + names, xerr := repo.CampaignsUsingSegment(ctx, f.org, acme.ID) + if xerr != nil || len(names) != 1 || names[0] != "RevOps outreach" { + t.Fatalf("using = %v, %v", names, xerr) + } + + // Unlinking everything keeps the enrolled leads (alice, carol, bob). + if xerr := repo.SetForCampaign(ctx, f.org, f.other, []uuid.UUID{}); xerr != nil { + t.Fatalf("unlink: %v", xerr) + } + if links, _ = repo.ListForCampaign(ctx, f.org, f.other); len(links) != 0 { + t.Fatalf("links after unlink = %+v", links) + } + inOther := models.SegmentCondition{Field: "campaign", Operator: "in", Values: []string{f.other.String()}} + if got := segCount(t, repo, f.org, models.SegmentMatchAll, inOther); got != 3 { + t.Errorf("leads after unlink = %d, want 3", got) + } +} diff --git a/web/src/components/app/contacts/ContactsTable.tsx b/web/src/components/app/contacts/ContactsTable.tsx index 89b284a90..a048073f0 100644 --- a/web/src/components/app/contacts/ContactsTable.tsx +++ b/web/src/components/app/contacts/ContactsTable.tsx @@ -36,6 +36,7 @@ import { SparklesIcon, TrashIcon, UploadIcon, + UserMinusIcon, UserPlusIcon, UsersIcon, XIcon, @@ -75,8 +76,9 @@ import AddToSegmentMenu from "@/components/app/segments/AddToSegmentMenu"; import SegmentEditor from "@/components/app/segments/SegmentEditor"; import { filtersToSegment } from "@/components/app/segments/filtersToSegment"; import type { SegmentCondition } from "@/lib/api/models/app/segments/Segment"; -import AddSegmentLeadsDialog from "@/components/app/segments/AddSegmentLeadsDialog"; -import { useSetSegmentMembers } from "@/lib/api/hooks/app/segments"; +import CampaignSegmentsDialog from "@/components/app/segments/CampaignSegmentsDialog"; +import { useCampaignSegments, useSetSegmentMembers } from "@/lib/api/hooks/app/segments"; +import useUpdateContactsBulk from "@/lib/api/hooks/app/contacts/useUpdateContactsBulk"; import useAiMetered from "@/hooks/useAiMetered"; import SyncSourcesPanel from "./SyncSourcesPanel"; import { CategoryChip } from "./CategoryPicker"; @@ -178,6 +180,26 @@ export default function ContactsTable({ } } const contactsBulkDelete = useDeleteContacts(); + const bulkUpdate = useUpdateContactsBulk(); + // Linked segments (live audience) for the toolbar badge on a Leads tab. + const campaignSegments = useCampaignSegments(current_campaign?.id, !!current_campaign); + + // In a campaign, "remove" detaches the leads; the contacts themselves stay. + async function removeFromCampaign(ids: string[]) { + if (!current_campaign || ids.length === 0 || bulkUpdate.isPending) return; + try { + await bulkUpdate.mutateAsync({ + contacts: ids, + add_campaigns: [], + remove_campaigns: [current_campaign.id], + fields: [], + }); + toast.success(`Removed ${ids.length} lead${ids.length === 1 ? "" : "s"} from ${current_campaign.name}`); + setSelected((bef) => bef.filter((x) => !ids.includes(x))); + } catch (err) { + toast.error(buildError(err as AppError)); + } + } // Connected CRM targets the "Push to CRM" bulk action can reach. Driven by // the org's live connections (backend enforces the push permission). @@ -363,6 +385,15 @@ export default function ContactsTable({ await bulkDelete(); }) } + onRemoveFromCampaign={ + embedded + ? (id) => + confirm?.show( + "Remove this lead from the campaign? The contact stays in your workspace.", + async () => removeFromCampaign([id]), + ) + : undefined + } emptyTitle={ subFilter !== "all" ? `No ${subFilter} contacts` @@ -403,7 +434,7 @@ export default function ContactsTable({ icon={} onClick={() => campaignWrite.guard(() => setFromSegmentOpen(true))({})} > - From segment + Link a segment + } + onClick={() => campaignWrite.guard(() => setFromSegmentOpen(true))({})} + > + Segments + {(campaignSegments.data?.length ?? 0) > 0 ? ` (${campaignSegments.data?.length})` : ""} + } @@ -508,6 +547,14 @@ export default function ContactsTable({ segment={segment} onExclude={excludeFromSegment} excluding={segmentMembers.isPending} + campaign={current_campaign} + onRemoveFromCampaign={() => + confirm?.show( + `Remove ${selected.length} lead${selected.length === 1 ? "" : "s"} from this campaign? The contacts stay in your workspace.`, + async () => removeFromCampaign(selected), + ) + } + removing={bulkUpdate.isPending} /> setFromContactsOpen(false)} campaign={current_campaign} /> - setFromSegmentOpen(false)} campaign={current_campaign} @@ -780,6 +827,7 @@ function ContactsTableBody({ onToggleAll, onRowClick, onDelete, + onRemoveFromCampaign, emptyTitle, emptyBody, emptyCta, @@ -818,6 +866,9 @@ function ContactsTableBody({ onToggleAll: () => void; onRowClick: (id: string, tab?: ContactSlideTab) => void; onDelete: (id: string) => void; + // In a campaign, the row's destructive action detaches the lead instead + // of deleting the contact from the whole workspace. + onRemoveFromCampaign?: (id: string) => void; emptyTitle: string; emptyBody: string; emptyCta: React.ReactNode; @@ -1078,14 +1129,25 @@ function ContactsTableBody({ e.stopPropagation()}> {/* Touch-safe: always visible on mobile, hover-reveal on desktop. */}
- + {onRemoveFromCampaign ? ( + + ) : ( + + )} )} + {campaign && onRemoveFromCampaign && ( + + )} +
+

+ Contacts in a linked segment become leads automatically, now and whenever the segment grows. + Removing a link stops new enrolment; existing leads stay. +

+
- {segments.isPending ? ( + {loading ? (
{[...Array(4)].map((_, i) => (
@@ -133,12 +176,12 @@ export default function AddSegmentLeadsDialog({ ) : (
    {list.map((s) => { - const on = picked === s.id; + const on = picked.has(s.id); return (
- Adds today's members. Existing leads are skipped. + + {picked.size === 0 ? "No segments linked" : `${picked.size} segment${picked.size === 1 ? "" : "s"} linked`} +
diff --git a/web/src/lib/api/client/app/contacts/importContacts.ts b/web/src/lib/api/client/app/contacts/importContacts.ts index ee3a5bbc8..5e2a5247c 100644 --- a/web/src/lib/api/client/app/contacts/importContacts.ts +++ b/web/src/lib/api/client/app/contacts/importContacts.ts @@ -46,6 +46,7 @@ export interface ImportCommitOptions { has_header: boolean; category_ids?: string[]; campaign_ids?: string[]; + segment_ids?: string[]; subscribed_default?: boolean; } diff --git a/web/src/lib/api/client/app/segments/index.ts b/web/src/lib/api/client/app/segments/index.ts index 4b28db353..c315c016a 100644 --- a/web/src/lib/api/client/app/segments/index.ts +++ b/web/src/lib/api/client/app/segments/index.ts @@ -1,6 +1,7 @@ import Request from "../../Request"; import type Segment from "@/lib/api/models/app/segments/Segment"; import type { + CampaignSegmentLink, ContactSegment, SegmentOverride, SegmentAddToCampaignResult, @@ -71,6 +72,27 @@ export async function listSegmentOverrides(id: string): Promise { + const res = await Request<{ data: CampaignSegmentLink[] }>({ + method: "GET", + url: `/campaigns/${campaignId}/segments`, + authorization: true, + }); + return res.data ?? []; +} + +export async function setCampaignSegments( + campaignId: string, + segmentIds: string[], +): Promise<{ data: CampaignSegmentLink[]; added: number }> { + return await Request<{ data: CampaignSegmentLink[]; added: number }>({ + method: "PUT", + url: `/campaigns/${campaignId}/segments`, + data: { segment_ids: segmentIds }, + authorization: true, + }); +} + export async function addSegmentToCampaign(id: string, campaignId: string): Promise { return await Request({ method: "POST", diff --git a/web/src/lib/api/hooks/app/segments/index.ts b/web/src/lib/api/hooks/app/segments/index.ts index 4805026e5..3f5f82405 100644 --- a/web/src/lib/api/hooks/app/segments/index.ts +++ b/web/src/lib/api/hooks/app/segments/index.ts @@ -1,6 +1,8 @@ import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { addSegmentToCampaign, + listCampaignSegments, + setCampaignSegments, listContactSegments, listSegmentOverrides, createSegment, @@ -107,6 +109,30 @@ export function useSetSegmentMembers() { }); } +// Keyed under ["campaigns", id, ...] so the campaign spine refreshes it. +export function useCampaignSegments(campaignId: string | undefined, enabled = true) { + return useQuery({ + queryKey: ["campaigns", campaignId, "segments"], + queryFn: () => listCampaignSegments(campaignId as string), + enabled: enabled && !!campaignId, + }); +} + +export function useSetCampaignSegments() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: ({ campaignId, segmentIds }: { campaignId: string; segmentIds: string[] }) => + setCampaignSegments(campaignId, segmentIds), + // Linking enrols leads right away, so the campaign's contact list moves. + onSuccess: (_res, vars) => + Promise.all([ + queryClient.invalidateQueries({ queryKey: ["campaigns", vars.campaignId, "segments"] }), + queryClient.invalidateQueries({ queryKey: ["contacts"] }), + queryClient.invalidateQueries({ queryKey: ["segments"] }), + ]), + }); +} + export function useAddSegmentToCampaign() { const queryClient = useQueryClient(); return useMutation({ diff --git a/web/src/lib/api/models/app/segments/Segment.ts b/web/src/lib/api/models/app/segments/Segment.ts index 91c47b697..2ec1064bb 100644 --- a/web/src/lib/api/models/app/segments/Segment.ts +++ b/web/src/lib/api/models/app/segments/Segment.ts @@ -83,6 +83,16 @@ export interface SegmentAddToCampaignResult { members: number; } +// One segment linked to a campaign as a live audience source. +export interface CampaignSegmentLink { + segment_id: string; + name: string; + color: string; + description: string; + contact_count: number; + linked_at: string; +} + // Operators per field kind, mirrored from the backend catalog. export const SEGMENT_OPERATORS: Record = { text: [ From 8db032a6568308c073ba3fd3fcf545a8670097c5 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Tue, 1 Sep 2026 00:33:42 -0700 Subject: [PATCH 2/9] feat: address review on segment-campaign linking: a campaign_lead_removals table (migration 000115, orgtransfer-registered) records hand-removed leads so the automatic segment sync never re-adds them while a manual add or the one-shot enrol clears the record, PUT /campaigns/:id/segments rejects an omitted segment_ids so {} cannot detach everything, rows.Err checks on the four new link queries so a truncated read cannot commit as success, write-path enrolment syncs detached from the request goroutine with a per-org in-flight dedupe, per-campaign 30s deadlines in the sweep instead of one shared budget, and an error-with-retry state in the linked-segments dialog --- docs/content/docs/api/reference/campaigns.mdx | 2 +- docs/content/docs/guides/campaigns.mdx | 2 +- docs/content/docs/guides/segments.mdx | 2 +- internal/api/handler/segment.go | 6 ++ internal/app/orgtransfer/spec.go | 5 ++ internal/app/segment/service.go | 64 ++++++++++++++----- .../000115_campaign_lead_removals.down.sql | 1 + .../000115_campaign_lead_removals.up.sql | 10 +++ internal/repository/pg_contact.go | 59 ++++++++++++++--- internal/repository/pg_segment.go | 44 +++++++++++-- internal/repository/segment_live_test.go | 32 ++++++++++ .../app/segments/CampaignSegmentsDialog.tsx | 18 ++++++ 12 files changed, 210 insertions(+), 35 deletions(-) create mode 100644 internal/infrastructure/db/migrations/000115_campaign_lead_removals.down.sql create mode 100644 internal/infrastructure/db/migrations/000115_campaign_lead_removals.up.sql diff --git a/docs/content/docs/api/reference/campaigns.mdx b/docs/content/docs/api/reference/campaigns.mdx index a7cb7690c..3148172fd 100644 --- a/docs/content/docs/api/reference/campaigns.mdx +++ b/docs/content/docs/api/reference/campaigns.mdx @@ -808,7 +808,7 @@ A `data` array of link objects (no pagination wrapper). `PUT /campaigns/:id/segments` -Atomically replace the campaign's linked segments with the supplied set (up to 20). Every current member of a newly linked segment is enrolled as a lead immediately, and contacts who enter a linked segment later are enrolled automatically, within about 2 minutes. Enrolment is additive: a contact who leaves a segment keeps their lead row, and removing a segment from the set stops future enrolment without touching existing leads. An active campaign is woken to send to the new leads; a completed campaign is restarted through the full launch checks when a linked segment grows. A linked segment cannot be deleted (`DELETE /segments/:id` returns `409`) until it is removed here. **Scope** `WRITE_CAMPAIGNS` · **Org permission** `manage_campaigns`. +Atomically replace the campaign's linked segments with the supplied set (up to 20). Every current member of a newly linked segment is enrolled as a lead immediately, and contacts who enter a linked segment later are enrolled automatically, within about 2 minutes. Enrolment is additive: a contact who leaves a segment keeps their lead row, and removing a segment from the set stops future enrolment without touching existing leads. A lead removed from the campaign by hand is never re-added automatically; a manual add (or the one-shot enrol below) clears that removal record. Omitting `segment_ids` returns `400`; send an explicit empty array to detach every segment. An active campaign is woken to send to the new leads; a completed campaign is restarted through the full launch checks when a linked segment grows. A linked segment cannot be deleted (`DELETE /segments/:id` returns `409`) until it is removed here. **Scope** `WRITE_CAMPAIGNS` · **Org permission** `manage_campaigns`. | Parameter | In | Type | Description | | --- | --- | --- | --- | diff --git a/docs/content/docs/guides/campaigns.mdx b/docs/content/docs/guides/campaigns.mdx index 815cc08b0..13bc4cc11 100644 --- a/docs/content/docs/guides/campaigns.mdx +++ b/docs/content/docs/guides/campaigns.mdx @@ -63,7 +63,7 @@ The **Leads** tab takes contacts four ways. **From contacts** opens a picker ove **Segments** on the same toolbar links [segments](/guides/segments/) to the campaign as a live audience, up to 20 per campaign. Linking enrols every current member immediately, and contacts who enter a linked segment later are enrolled on their own, within a couple of minutes. An active campaign wakes to send to them, a finished one restarts through the usual launch checks, and a paused one accumulates them for later. Enrolment is additive: a contact who leaves the segment keeps their lead row, and detaching a segment stops future enrolment without touching existing leads. See [linking a segment to a campaign](/guides/segments/#linking-a-segment-to-a-campaign). -A lead can also be taken out again: the remove button on a lead's row, or **Remove from campaign** in the selection bar when several are ticked, drops the lead from this campaign without deleting the contact. +A lead can also be taken out again: the remove button on a lead's row, or **Remove from campaign** in the selection bar when several are ticked, drops the lead from this campaign without deleting the contact. The removal sticks: a linked segment will not re-enrol that contact automatically, even while they still match it, until you add them back yourself. ### Lead statuses diff --git a/docs/content/docs/guides/segments.mdx b/docs/content/docs/guides/segments.mdx index b9685ff0d..cd47b7240 100644 --- a/docs/content/docs/guides/segments.mdx +++ b/docs/content/docs/guides/segments.mdx @@ -59,7 +59,7 @@ Where **Add to campaign** copies today's members once, a linked segment is a liv Linking enrols every current member as a lead immediately. From then on, any contact who enters the segment, a new contact, an import, a condition edit, a manual pin-in, or simply drifting into a date or engagement condition, is enrolled automatically; a background check picks up drift about every 2 minutes. -Enrolment is additive. A contact who later falls out of the segment keeps their lead row and their progress; the link only ever adds. Detaching a segment stops future enrolment but leaves the leads it already added in place. +Enrolment is additive. A contact who later falls out of the segment keeps their lead row and their progress; the link only ever adds. Detaching a segment stops future enrolment but leaves the leads it already added in place. Removing a lead from the campaign by hand is respected too: automatic enrolment will not re-add that contact, even while they still match a linked segment, until you add them back yourself. The campaign reacts to growth by status: diff --git a/internal/api/handler/segment.go b/internal/api/handler/segment.go index 058cc6ead..a565527f0 100644 --- a/internal/api/handler/segment.go +++ b/internal/api/handler/segment.go @@ -248,6 +248,12 @@ func (h *Handler) SetCampaignSegments(c *gin.Context) { errx.Handle(c, errx.ErrInvalid) return } + // An omitted field must not read as "detach everything"; only an + // explicit [] does that. + if in.SegmentIDs == nil { + errx.Handle(c, errx.New(errx.BadRequest, "segment_ids is required; send [] to detach all segments")) + return + } links, added, xerr := h.SegmentService.SetCampaignSegments(c.Request.Context(), orgID, campaignID, &in) if xerr != nil { errx.Handle(c, xerr) diff --git a/internal/app/orgtransfer/spec.go b/internal/app/orgtransfer/spec.go index 32d90a714..f2a9f544e 100644 --- a/internal/app/orgtransfer/spec.go +++ b/internal/app/orgtransfer/spec.go @@ -352,6 +352,11 @@ var Tables = []Table{ Name: "campaign_segments", Group: models.OrgDataGroupCampaigns, Scope: `campaign_id IN ` + orgCampaigns, }, + { + Name: "campaign_lead_removals", Group: models.OrgDataGroupCampaigns, + Scope: `campaign_id IN ` + orgCampaigns, + Note: "Must travel, or linked segments on the destination re-add every lead the user removed by hand.", + }, { Name: "campaign_ab_assignments", Group: models.OrgDataGroupCampaigns, Scope: `campaign_id IN ` + orgCampaigns, diff --git a/internal/app/segment/service.go b/internal/app/segment/service.go index 5d6e69469..026134178 100644 --- a/internal/app/segment/service.go +++ b/internal/app/segment/service.go @@ -8,6 +8,7 @@ import ( "context" "fmt" "strings" + "sync" "time" "github.com/google/uuid" @@ -68,6 +69,9 @@ type service struct { fields CustomFieldLister waker CampaignWaker starter CampaignStarter + // orgSyncInFlight dedupes concurrent org-wide enrolment passes, keyed by + // org id. + orgSyncInFlight sync.Map } func NewService(repo repository.SegmentRepository, fields CustomFieldLister) Service { @@ -388,25 +392,44 @@ func (s *service) syncLinkedCampaign(ctx context.Context, lc models.LinkedCampai // syncLinkedCampaignsForSegments re-enrols the campaigns linked to any of the // given segments. Nested references (a linked segment built on this one) are -// not chased here; the periodic sweep covers them. +// not chased here; the periodic sweep covers them. Detached from the caller's +// request: the enrolment scans grow with the contact list, and this is +// declared best effort with the sweep as the backstop. func (s *service) syncLinkedCampaignsForSegments(ctx context.Context, orgID uuid.UUID, segmentIDs []uuid.UUID) { - links, xerr := s.repo.LinkedCampaignsForSegments(ctx, orgID, segmentIDs) - if xerr != nil { - return - } - for _, lc := range links { - s.syncLinkedCampaign(ctx, lc) - } + bg := context.WithoutCancel(ctx) + go func() { + rctx, cancel := context.WithTimeout(bg, 2*time.Minute) + defer cancel() + links, xerr := s.repo.LinkedCampaignsForSegments(rctx, orgID, segmentIDs) + if xerr != nil { + return + } + for _, lc := range links { + s.syncLinkedCampaign(rctx, lc) + } + }() } func (s *service) SyncOrgLinkedCampaigns(ctx context.Context, orgID uuid.UUID) { - links, xerr := s.repo.LinkedCampaigns(ctx, &orgID) - if xerr != nil { + // One in-flight pass per org: every contact write calls this, and a burst + // of writes must not stack org-wide enrolment scans. A write landing while + // a pass runs is picked up by the sweep within its interval. + if _, running := s.orgSyncInFlight.LoadOrStore(orgID, struct{}{}); running { return } - for _, lc := range links { - s.syncLinkedCampaign(ctx, lc) - } + bg := context.WithoutCancel(ctx) + go func() { + defer s.orgSyncInFlight.Delete(orgID) + rctx, cancel := context.WithTimeout(bg, 2*time.Minute) + defer cancel() + links, xerr := s.repo.LinkedCampaigns(rctx, &orgID) + if xerr != nil { + return + } + for _, lc := range links { + s.syncLinkedCampaign(rctx, lc) + } + }() } func (s *service) StartCampaignSegmentSync(ctx context.Context, interval time.Duration) { @@ -427,16 +450,23 @@ func (s *service) StartCampaignSegmentSync(ctx context.Context, interval time.Du } func (s *service) sweepLinkedCampaigns(ctx context.Context) { - rctx, cancel := context.WithTimeout(ctx, 2*time.Minute) - defer cancel() - links, xerr := s.repo.LinkedCampaigns(rctx, nil) + scanCtx, scanCancel := context.WithTimeout(ctx, 30*time.Second) + links, xerr := s.repo.LinkedCampaigns(scanCtx, nil) + scanCancel() if xerr != nil { log.Warn().Str("error", xerr.Message).Msg("segment sync: sweep scan failed") return } total := 0 + // Each campaign gets its own deadline: one shared budget would starve the + // same tail campaigns every pass once the instance holds enough links. for _, lc := range links { - total += s.syncLinkedCampaign(rctx, lc) + if ctx.Err() != nil { + return + } + cctx, cancel := context.WithTimeout(ctx, 30*time.Second) + total += s.syncLinkedCampaign(cctx, lc) + cancel() } if total > 0 { log.Info().Int("added", total).Msg("segment sync: sweep enrolled new leads") diff --git a/internal/infrastructure/db/migrations/000115_campaign_lead_removals.down.sql b/internal/infrastructure/db/migrations/000115_campaign_lead_removals.down.sql new file mode 100644 index 000000000..07c4647d9 --- /dev/null +++ b/internal/infrastructure/db/migrations/000115_campaign_lead_removals.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS campaign_lead_removals; diff --git a/internal/infrastructure/db/migrations/000115_campaign_lead_removals.up.sql b/internal/infrastructure/db/migrations/000115_campaign_lead_removals.up.sql new file mode 100644 index 000000000..74232fbe9 --- /dev/null +++ b/internal/infrastructure/db/migrations/000115_campaign_lead_removals.up.sql @@ -0,0 +1,10 @@ +-- A lead a user removed from a campaign by hand. Automatic segment enrolment +-- must not re-add these pairs; an explicit manual add clears the record. +CREATE TABLE campaign_lead_removals ( + campaign_id uuid NOT NULL REFERENCES campaigns (id) ON DELETE CASCADE, + contact_id uuid NOT NULL REFERENCES contacts (id) ON DELETE CASCADE, + created_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (campaign_id, contact_id) +); + +CREATE INDEX idx_campaign_lead_removals_contact ON campaign_lead_removals (contact_id); diff --git a/internal/repository/pg_contact.go b/internal/repository/pg_contact.go index 0505b4d38..267581ffe 100644 --- a/internal/repository/pg_contact.go +++ b/internal/repository/pg_contact.go @@ -1828,14 +1828,23 @@ func (r *contactRepository) Update(ctx context.Context, userID, contactID string toDelete := utils.Difference(currentCampaignIDs, wantIDs) if len(toDelete) > 0 { + // Record the removal so linked-segment enrolment does not re-add + // the pair; a later manual add clears it again. query = ` - DELETE FROM campaign_leads cl - USING campaigns cam - WHERE cl.contact_id = $1 - AND cl.campaign_id = cam.id - AND cam.id = ANY($2::uuid[]) - AND cam.organization_id = $3 - RETURNING cl.campaign_id + WITH gone AS ( + DELETE FROM campaign_leads cl + USING campaigns cam + WHERE cl.contact_id = $1 + AND cl.campaign_id = cam.id + AND cam.id = ANY($2::uuid[]) + AND cam.organization_id = $3 + RETURNING cl.contact_id, cl.campaign_id + ), mark AS ( + INSERT INTO campaign_lead_removals (campaign_id, contact_id) + SELECT campaign_id, contact_id FROM gone + ON CONFLICT (campaign_id, contact_id) DO UPDATE SET created_at = now() + ) + SELECT campaign_id FROM gone ` params := []any{contactID, toDelete, orgID} rows, err := tx.Query(ctx, query, params...) @@ -1853,6 +1862,14 @@ func (r *contactRepository) Update(ctx context.Context, userID, contactID string if len(toInsert) > 0 { query = ` + WITH cleared AS ( + DELETE FROM campaign_lead_removals r + USING campaigns cam + WHERE r.contact_id = $1 + AND r.campaign_id = cam.id + AND cam.id = ANY($2::uuid[]) + AND cam.organization_id = $3 + ) INSERT INTO campaign_leads (contact_id, campaign_id) SELECT $1, cam.id FROM campaigns cam @@ -2109,7 +2126,11 @@ func (r *contactRepository) BulkUpdate(ctx context.Context, userID string, orgID return nil } if len(data.RemoveCampaigns) > 0 { - if xerr := link(`DELETE FROM campaign_leads cl + // Only pairs that actually held a lead are recorded as removals, so + // the automatic segment enrolment never re-adds a hand-removed lead + // while contacts merely listed in the request stay eligible. + if xerr := link(`WITH gone AS ( + DELETE FROM campaign_leads cl USING contacts c, campaigns cam WHERE cl.contact_id = c.id AND cl.campaign_id = cam.id @@ -2117,14 +2138,32 @@ func (r *contactRepository) BulkUpdate(ctx context.Context, userID string, orgID AND cam.organization_id = $1 AND cl.contact_id = ANY($2) AND cl.campaign_id = ANY($3) - RETURNING cl.contact_id, cl.campaign_id`, + RETURNING cl.contact_id, cl.campaign_id + ), mark AS ( + INSERT INTO campaign_lead_removals (campaign_id, contact_id) + SELECT campaign_id, contact_id FROM gone + ON CONFLICT (campaign_id, contact_id) DO UPDATE SET created_at = now() + ) + SELECT contact_id, campaign_id FROM gone`, models.ActivityCampaignRemoved, logCampaignLinks, orgID, data.Contacts, data.RemoveCampaigns); xerr != nil { return nil, xerr } } if len(data.AddCampaigns) > 0 { - if xerr := link(`INSERT INTO campaign_leads (contact_id, campaign_id) + // A manual add clears the removal record: the user changed their mind, + // so linked segments may manage this pair again. + if xerr := link(`WITH cleared AS ( + DELETE FROM campaign_lead_removals r + USING contacts c, campaigns cam + WHERE r.contact_id = c.id + AND r.campaign_id = cam.id + AND c.organization_id = $1 + AND cam.organization_id = $1 + AND r.contact_id = ANY($2) + AND r.campaign_id = ANY($3::uuid[]) + ) + INSERT INTO campaign_leads (contact_id, campaign_id) SELECT c.id, cam.id FROM contacts c CROSS JOIN campaigns cam diff --git a/internal/repository/pg_segment.go b/internal/repository/pg_segment.go index 22f1a11a4..ba3db5b35 100644 --- a/internal/repository/pg_segment.go +++ b/internal/repository/pg_segment.go @@ -299,7 +299,7 @@ func (r *segmentRepository) AddToCampaign(ctx context.Context, orgID uuid.UUID, return nil, errx.InternalError() } - links, err := insertSegmentLeads(ctx, tx, orgID, actorID(actor), clause, args, campaignID) + links, err := insertSegmentLeads(ctx, tx, orgID, actorID(actor), clause, args, campaignID, false) if err != nil { db.CaptureError(err, "segment enrol", nil, "query") return nil, errx.InternalError() @@ -314,12 +314,26 @@ func (r *segmentRepository) AddToCampaign(ctx context.Context, orgID uuid.UUID, // insertSegmentLeads enrols every contact matching the precompiled segment // clause as a lead, logging a campaign_added activity for each row that was // actually new. The campaign is bound after the clause's own parameters. -func insertSegmentLeads(ctx context.Context, tx pgx.Tx, orgID uuid.UUID, actor *uuid.UUID, clause string, args []any, campaignID uuid.UUID) ([]contactLink, error) { +// +// respectRemovals decides what a manual "remove from campaign" means here: +// the automatic sync honours the removal record and skips the pair, while an +// explicit enrol (the one-shot add-to-campaign) clears it and re-adds. +func insertSegmentLeads(ctx context.Context, tx pgx.Tx, orgID uuid.UUID, actor *uuid.UUID, clause string, args []any, campaignID uuid.UUID, respectRemovals bool) ([]contactLink, error) { args = append(args, campaignID) + guard := "" + if respectRemovals { + guard = fmt.Sprintf(` AND NOT EXISTS (SELECT 1 FROM campaign_lead_removals r WHERE r.campaign_id = $%d AND r.contact_id = c.id)`, len(args)) + } else { + clearQ := fmt.Sprintf(`DELETE FROM campaign_lead_removals r + WHERE r.campaign_id = $%d AND r.contact_id IN (SELECT c.id FROM contacts c WHERE c.organization_id = $1 AND (%s))`, len(args), clause) + if _, err := tx.Exec(ctx, clearQ, args...); err != nil { + return nil, err + } + } insertQ := fmt.Sprintf(`INSERT INTO campaign_leads (contact_id, campaign_id) - SELECT c.id, $%d::uuid FROM contacts c WHERE c.organization_id = $1 AND (%s) + SELECT c.id, $%d::uuid FROM contacts c WHERE c.organization_id = $1 AND (%s)%s ON CONFLICT DO NOTHING - RETURNING contact_id, campaign_id`, len(args), clause) + RETURNING contact_id, campaign_id`, len(args), clause, guard) rows, err := tx.Query(ctx, insertQ, args...) if err != nil { return nil, err @@ -377,6 +391,12 @@ func (r *segmentRepository) ListForCampaign(ctx context.Context, orgID, campaign matches = append(matches, models.SegmentMatch(match)) conds = append(conds, cs) } + // A mid-stream read failure ends Next() early with no scan error; without + // this the Leads tab would render a truncated link list as the truth. + if err := rows.Err(); err != nil { + db.CaptureError(err, "campaign segments list", nil, "rows") + return nil, errx.InternalError() + } for i := range out { n, xerr := r.Count(ctx, orgID, &out[i].SegmentID, matches[i], conds[i]) if xerr != nil { @@ -461,6 +481,12 @@ func (r *segmentRepository) SyncCampaignSegments(ctx context.Context, orgID, cam segmentIDs = append(segmentIDs, id) } rows.Close() + // A truncated id list here would enrol part of the audience and still + // commit as a success, so a read failure has to abort the sync. + if err := rows.Err(); err != nil { + db.CaptureError(err, "campaign segments sync", nil, "rows") + return 0, errx.InternalError() + } total := 0 for _, segID := range segmentIDs { @@ -474,7 +500,7 @@ func (r *segmentRepository) SyncCampaignSegments(ctx context.Context, orgID, cam if clause == "FALSE" { continue } - links, lerr := insertSegmentLeads(ctx, tx, orgID, nil, clause, args, campaignID) + links, lerr := insertSegmentLeads(ctx, tx, orgID, nil, clause, args, campaignID, true) if lerr != nil { db.CaptureError(lerr, "segment enrol", nil, "query") return 0, errx.InternalError() @@ -505,6 +531,10 @@ func scanLinkedCampaigns(rows pgx.Rows) ([]models.LinkedCampaign, *errx.Error) { } out = append(out, l) } + if err := rows.Err(); err != nil { + db.CaptureError(err, "linked campaigns", nil, "rows") + return nil, errx.InternalError() + } return out, nil } @@ -555,6 +585,10 @@ func (r *segmentRepository) CampaignsUsingSegment(ctx context.Context, orgID, se } names = append(names, n) } + if err := rows.Err(); err != nil { + db.CaptureError(err, "campaigns using segment", nil, "rows") + return nil, errx.InternalError() + } return names, nil } diff --git a/internal/repository/segment_live_test.go b/internal/repository/segment_live_test.go index 0d0808b84..c45ffeac8 100644 --- a/internal/repository/segment_live_test.go +++ b/internal/repository/segment_live_test.go @@ -286,6 +286,38 @@ func TestLiveSegmentCampaignLinks(t *testing.T) { t.Fatalf("post-include sync = %d, want 1", added) } + // A hand-removed lead stays removed: the sync skips the pair until a + // manual add clears the record, and the explicit one-shot enrol overrides + // and clears it too. + handle, _ := liveContactDB(t) + contacts := NewContactRepostory(handle) + removeBob := &models.BulkEditContactsData{Contacts: []string{f.bob.String()}, RemoveCampaigns: []string{f.other.String()}} + if _, xerr := contacts.BulkUpdate(ctx, f.owner.String(), f.org, removeBob); xerr != nil { + t.Fatalf("remove: %v", xerr) + } + if added, _ = repo.SyncCampaignSegments(ctx, f.org, f.other); added != 0 { + t.Fatalf("sync after manual removal = %d, want 0", added) + } + if _, xerr := contacts.BulkUpdate(ctx, f.owner.String(), f.org, &models.BulkEditContactsData{ + Contacts: []string{f.bob.String()}, AddCampaigns: []string{f.other.String()}, + }); xerr != nil { + t.Fatalf("re-add: %v", xerr) + } + var removals int + if err := handle.QueryRow(ctx, `SELECT COUNT(*) FROM campaign_lead_removals WHERE campaign_id = $1`, f.other).Scan(&removals); err != nil || removals != 0 { + t.Fatalf("removals after manual re-add = %d, %v", removals, err) + } + if _, xerr := contacts.BulkUpdate(ctx, f.owner.String(), f.org, removeBob); xerr != nil { + t.Fatalf("remove again: %v", xerr) + } + out, xerr := repo.AddToCampaign(ctx, f.org, f.owner.String(), acme.ID, f.other) + if xerr != nil || out.Added != 1 { + t.Fatalf("one-shot after removal = %+v, %v", out, xerr) + } + if err := handle.QueryRow(ctx, `SELECT COUNT(*) FROM campaign_lead_removals WHERE campaign_id = $1`, f.other).Scan(&removals); err != nil || removals != 0 { + t.Fatalf("removals after one-shot = %d, %v", removals, err) + } + // The sweep and the targeted lookup both see the linked campaign, and the // delete guard reports it by name. linked, xerr := repo.LinkedCampaigns(ctx, &f.org) diff --git a/web/src/components/app/segments/CampaignSegmentsDialog.tsx b/web/src/components/app/segments/CampaignSegmentsDialog.tsx index 7b0f57293..3bfe395d4 100644 --- a/web/src/components/app/segments/CampaignSegmentsDialog.tsx +++ b/web/src/components/app/segments/CampaignSegmentsDialog.tsx @@ -108,6 +108,12 @@ export default function CampaignSegmentsDialog({ } const loading = segments.isPending || (linked.isPending && !seeded); + // A failed load must not strand the dialog as "loaded but Save disabled". + const loadError = segments.isError || (linked.isError && !seeded); + const retryLoad = () => { + if (segments.isError) void segments.refetch(); + if (linked.isError) void linked.refetch(); + }; return ( @@ -168,6 +174,18 @@ export default function CampaignSegmentsDialog({
))}
+ ) : loadError ? ( +
+

Couldn't load segments

+

Check your connection and try again.

+ +
) : list.length === 0 ? (

{query ? "No segments match" : "No segments yet"}

From 7b6830daa563d2500bc78510f9460dd2384f18cf Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Tue, 1 Sep 2026 01:32:56 -0700 Subject: [PATCH 3/9] feat: say segments out loud in the API scope catalog: the READ_CONTACTS and WRITE_CONTACTS descriptions served to the API key dialog (and the docs permission table) now name segments, since segment CRUD, previews, member overrides and enrolment ride on the contact scopes and the picker never said so --- docs/content/docs/api/permissions.mdx | 4 ++-- internal/models/api_permission.go | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/docs/content/docs/api/permissions.mdx b/docs/content/docs/api/permissions.mdx index 0409992ca..f0ca6bbc6 100644 --- a/docs/content/docs/api/permissions.mdx +++ b/docs/content/docs/api/permissions.mdx @@ -13,12 +13,12 @@ These same permissions are the [OAuth](/api/oauth/) scopes, lowercased: `READ_EM |------------|-----|-------|----------|-------------| | `READ_EMAILS` | 0 | 1 | read | View email accounts and settings | | `READ_CAMPAIGNS` | 1 | 2 | read | View campaigns and sequences | -| `READ_CONTACTS` | 2 | 4 | read | View contact lists, notes, and activities | +| `READ_CONTACTS` | 2 | 4 | read | View contact lists, segments, notes, and activities | | `READ_UNIBOX` | 3 | 8 | read | Access unified inbox | | `READ_ANALYTICS` | 4 | 16 | read | View analytics and statistics | | `WRITE_EMAILS` | 5 | 32 | write | Modify email account settings | | `WRITE_CAMPAIGNS` | 6 | 64 | write | Create and modify campaigns and sequences | -| `WRITE_CONTACTS` | 7 | 128 | write | Create and modify contacts, notes, activities | +| `WRITE_CONTACTS` | 7 | 128 | write | Create and modify contacts, segments, notes, activities | | `WRITE_UNIBOX` | 8 | 256 | write | Mark emails as read/unread and send replies | | `BULK_CONTACTS` | 9 | 512 | bulk | Bulk import/export/delete contacts | | `BULK_CAMPAIGNS` | 10 | 1024 | bulk | Bulk campaign operations | diff --git a/internal/models/api_permission.go b/internal/models/api_permission.go index 04730b9e2..f0f54fb62 100644 --- a/internal/models/api_permission.go +++ b/internal/models/api_permission.go @@ -8,14 +8,14 @@ const ( // Read permissions APIPermReadEmails uint64 = 1 << iota // Read email accounts APIPermReadCampaigns // Read campaigns + sequences - APIPermReadContacts // Read contacts, notes, activities + APIPermReadContacts // Read contacts, segments, notes, activities APIPermReadUnibox // Read unified inbox APIPermReadAnalytics // Read analytics + statistics // Write permissions APIPermWriteEmails // Modify email accounts APIPermWriteCampaigns // Create/modify campaigns + sequences - APIPermWriteContacts // Create/modify contacts, notes, activities + APIPermWriteContacts // Create/modify contacts, segments, notes, activities APIPermWriteUnibox // Mark as seen, reply, etc. // Bulk operations (separate so a key can read+write without bulk power) @@ -98,7 +98,7 @@ type APIPermission struct { var AllAPIPermissions = []APIPermission{ {"READ_EMAILS", APIPermReadEmails, "View email accounts and settings", "read"}, {"READ_CAMPAIGNS", APIPermReadCampaigns, "View campaigns and sequences", "read"}, - {"READ_CONTACTS", APIPermReadContacts, "View contact lists, notes, and activities", "read"}, + {"READ_CONTACTS", APIPermReadContacts, "View contact lists, segments, notes, and activities", "read"}, {"READ_UNIBOX", APIPermReadUnibox, "Access unified inbox", "read"}, {"READ_ANALYTICS", APIPermReadAnalytics, "View analytics and statistics", "read"}, {"READ_TEMPLATES", APIPermReadTemplates, "View reply templates", "read"}, @@ -106,7 +106,7 @@ var AllAPIPermissions = []APIPermission{ {"READ_AUDIT_LOGS", APIPermReadAuditLogs, "View organization audit logs", "read"}, {"WRITE_EMAILS", APIPermWriteEmails, "Modify email account settings", "write"}, {"WRITE_CAMPAIGNS", APIPermWriteCampaigns, "Create and modify campaigns and sequences", "write"}, - {"WRITE_CONTACTS", APIPermWriteContacts, "Create and modify contacts, notes, and activities", "write"}, + {"WRITE_CONTACTS", APIPermWriteContacts, "Create and modify contacts, segments, notes, and activities", "write"}, {"WRITE_UNIBOX", APIPermWriteUnibox, "Mark emails as read/unread and send replies", "write"}, {"WRITE_TEMPLATES", APIPermWriteTemplates, "Create and modify reply templates", "write"}, {"WRITE_CRM", APIPermWriteCRM, "Create and modify pipelines, deals, and CRM tasks", "write"}, From 55307b0ad9f4f9cbeef4aedabb793dcee69e2bd4 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Tue, 1 Sep 2026 02:57:30 -0700 Subject: [PATCH 4/9] feat: fix the segment dialog UI and make a contact created inside a segment join it (issue #285): the add-to-campaign picker now shows a humanised, colour-toned campaign status instead of the raw PAUSED_NO_ACCOUNTS enum, its footer wraps instead of clipping the hint mid-sentence and the Add leads button no longer breaks across two lines, campaign status labels move to a shared components/app/campaigns/status module, the Segments, Categories and Segment pages plus every unmapped settings and onboarding route get a document title so they stop reading Page not found, and POST /contacts takes a segments array that pins the new contacts in as include overrides (validated before the write, best-effort after it) which the New contact dialog sends when opened from a segment page --- docs/content/docs/api/reference/contacts.mdx | 1 + docs/content/docs/guides/segments.mdx | 4 +- internal/app/contact/handler.go | 75 +++++++++++++++++++ internal/models/contact.go | 5 ++ web/src/app/app/campaigns/page.tsx | 47 ++---------- web/src/components/app/AddContacts.tsx | 3 + web/src/components/app/campaigns/status.ts | 49 ++++++++++++ .../components/app/contacts/ContactsTable.tsx | 2 +- .../app/contacts/NewContactDialog.tsx | 14 +++- .../segments/AddSegmentToCampaignDialog.tsx | 20 +++-- .../app/segments/CampaignSegmentsDialog.tsx | 8 +- web/src/hooks/useDocumentTitle.ts | 15 ++++ .../api/hooks/app/contacts/useAddContacts.ts | 5 +- 13 files changed, 195 insertions(+), 53 deletions(-) create mode 100644 web/src/components/app/campaigns/status.ts diff --git a/docs/content/docs/api/reference/contacts.mdx b/docs/content/docs/api/reference/contacts.mdx index 1053bdbab..1bdca2873 100644 --- a/docs/content/docs/api/reference/contacts.mdx +++ b/docs/content/docs/api/reference/contacts.mdx @@ -151,6 +151,7 @@ A JSON array of contact objects (at least one, up to the per-request maximum; an | `phone` | string | No | Phone number. | | `campaigns` | string[] | No | Campaign IDs to add the contact to. | | `categories` | string[] | No | Category IDs to assign. | +| `segments` | string[] | No | Segment IDs to pin the contact into, as a manual include override, so it belongs whether or not the conditions match it. An unknown id is rejected with `400` before any contact is written. | | `custom_fields` | object | No | String key/value custom fields. Keys may use letters, numbers, underscores, spaces, and dashes. | | `subscribed` | boolean | No | Marketing-consent flag. Omit it to let a new contact default to subscribed and an existing one keep whatever it already had. | | `verification_status` | string | No | A verdict you already hold for the address, in Warmbly's vocabulary (`valid`, `risky`, `invalid`, `unknown`) or any known service's (`ok`, `catch-all`, `do_not_mail`, `deliverable`, `ok_for_all`, ...). Stored as an imported verdict that the background check leaves alone. A value no known service writes is rejected with `unknown_verification_status`. | diff --git a/docs/content/docs/guides/segments.mdx b/docs/content/docs/guides/segments.mdx index cd47b7240..a79b614f0 100644 --- a/docs/content/docs/guides/segments.mdx +++ b/docs/content/docs/guides/segments.mdx @@ -42,7 +42,9 @@ Conditions decide membership, and two overrides sit on top of them: The segment header shows how many contacts are pinned in or out, and a **Pinned contacts** panel below it lists them; **Back to automatic** clears an override so the conditions decide again. A contact's own drawer has a **Segments** section that shows every segment, whether the contact is in it, and the same pin in, pin out and back-to-automatic controls. -Sequences can pin as well: the **Add to segment** and **Remove from segment** action steps apply the override to a contact as it moves through a campaign flow, so a positive reply can drop someone into a "warm" segment automatically. So can an import: the file import wizard's **Add to segments** picker on its Options step pins every contact in the file into the segments you choose, so a fresh list can land straight in the audience it was collected for. +**New contact** on a segment page pins as well: a contact created from inside a segment joins that segment, so it appears in the list you created it from even when the conditions do not describe it yet. + +Sequences can pin too: the **Add to segment** and **Remove from segment** action steps apply the override to a contact as it moves through a campaign flow, so a positive reply can drop someone into a "warm" segment automatically. So can an import: the file import wizard's **Add to segments** picker on its Options step pins every contact in the file into the segments you choose, so a fresh list can land straight in the audience it was collected for. ## Using a segment diff --git a/internal/app/contact/handler.go b/internal/app/contact/handler.go index 263e6f9c6..35af8668d 100644 --- a/internal/app/contact/handler.go +++ b/internal/app/contact/handler.go @@ -7,6 +7,7 @@ import ( "time" "github.com/google/uuid" + "github.com/rs/zerolog/log" "github.com/warmbly/warmbly/internal/config" "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/models" @@ -56,10 +57,18 @@ func (s *contactService) Add(ctx context.Context, userID string, orgID uuid.UUID return nil, xerr } + // Resolved before the write so an unknown segment is a 400 rather than a + // contact that exists but never reached the segment it was created in. + pins, xerr := s.segmentPins(ctx, orgID, contacts) + if xerr != nil { + return nil, xerr + } + created, xerr := s.contactRepository.Add(ctx, userID, orgID, contacts) if xerr != nil { return nil, xerr } + s.applySegmentPins(ctx, orgID, pins, created) s.publishContactsReload(ctx, userID, "contacts:add") var attached []string @@ -71,6 +80,72 @@ func (s *contactService) Add(ctx context.Context, userID string, orgID uuid.UUID return created, nil } +// segmentPins maps each target segment to the positions in `contacts` that +// asked for it. Existence is checked once for the whole batch, so N contacts +// naming the same segment cost one lookup. +func (s *contactService) segmentPins(ctx context.Context, orgID uuid.UUID, contacts []models.AddContact) (map[uuid.UUID][]int, *errx.Error) { + var raw []string + for i := range contacts { + raw = append(raw, contacts[i].Segments...) + } + if len(raw) == 0 { + return nil, nil + } + valid, xerr := s.parseSegmentIDs(ctx, orgID, raw) + if xerr != nil { + return nil, xerr + } + known := make(map[uuid.UUID]bool, len(valid)) + for _, id := range valid { + known[id] = true + } + + pins := make(map[uuid.UUID][]int, len(valid)) + for i := range contacts { + seen := make(map[uuid.UUID]bool, len(contacts[i].Segments)) + for _, r := range contacts[i].Segments { + id, err := uuid.Parse(strings.TrimSpace(r)) + if err != nil || !known[id] { + return nil, errx.New(errx.BadRequest, "invalid segment id") + } + if seen[id] { + continue + } + seen[id] = true + pins[id] = append(pins[id], i) + } + } + return pins, nil +} + +// applySegmentPins writes the include overrides for a batch that has just been +// created. The repository answers one row per input contact, in order, so +// position i of `created` is the contact that position i of the request asked +// to pin. Best effort like wakeCampaigns: the contacts are already stored. +func (s *contactService) applySegmentPins(ctx context.Context, orgID uuid.UUID, pins map[uuid.UUID][]int, created []models.Contact) { + if len(pins) == 0 || s.segmentLinker == nil { + return + } + for segmentID, positions := range pins { + ids := make([]uuid.UUID, 0, len(positions)) + for _, i := range positions { + if i < len(created) { + ids = append(ids, created[i].ID) + } + } + if len(ids) == 0 { + continue + } + if _, xerr := s.segmentLinker.SetMembers(ctx, orgID, segmentID, ids, models.SegmentMemberInclude); xerr != nil { + log.Warn(). + Str("organization_id", orgID.String()). + Str("segment_id", segmentID.String()). + Int("contacts", len(ids)). + Msg("could not pin the new contacts into their segment") + } + } +} + func (s *contactService) Search(ctx context.Context, orgID, cursor, category, limit string, filters models.SearchContacts) (*models.ContactsResult, *errx.Error) { cursorId, err := paging.DecodeCursor(cursor) if err != nil { diff --git a/internal/models/contact.go b/internal/models/contact.go index 1332845a5..e1ceb3ea4 100644 --- a/internal/models/contact.go +++ b/internal/models/contact.go @@ -486,6 +486,11 @@ type AddContact struct { Campaigns []string `json:"campaigns"` Categories []string `json:"categories"` + // Segments the new contact joins as an include override, so a contact + // created from inside a segment belongs to it whatever the conditions + // say. Unknown ids are a 400 before anything is written. + Segments []string `json:"segments,omitempty"` + CustomFields map[string]string `json:"custom_fields"` // VerificationStatus is a verdict the caller already holds for this diff --git a/web/src/app/app/campaigns/page.tsx b/web/src/app/app/campaigns/page.tsx index d05e16a5d..e89ae56c9 100644 --- a/web/src/app/app/campaigns/page.tsx +++ b/web/src/app/app/campaigns/page.tsx @@ -47,6 +47,11 @@ import { TopbarAction, } from "@/components/layout/Page"; import { SearchInput } from "@/components/ui/field"; +import { + CAMPAIGN_STATUS_LABEL, + campaignStatusBucket as statusBucket, + campaignStatusTone as statusTone, +} from "@/components/app/campaigns/status"; import { PopoverMenu, PopoverMenuContent, @@ -60,47 +65,11 @@ import { type StatusFilter = "all" | "active" | "paused" | "draft" | "completed"; type SortMode = "newest" | "oldest" | "name"; -// Collapse the backend's raw campaign statuses into the buckets the list -// filters/counts by. paused_no_accounts / paused_trial_expired are auto-pause -// variants, so they belong with "paused"; anything unknown reads as draft. -function statusBucket(s?: string): "active" | "paused" | "completed" | "draft" { - if (s === "active") return "active"; - if (s === "completed") return "completed"; - if (s && s.startsWith("paused")) return "paused"; - return "draft"; -} - // Per-state label + leading mark for a campaign row. "active" renders the // animated dot-grid loader; every other state is a 14px lucide icon so the -// fixed-width leading slot keeps each row's name aligned. -const STATUS_LABEL: Record = { - active: "running", - paused: "paused", - paused_no_accounts: "no accounts", - paused_trial_expired: "trial expired", - paused_guardrail: "auto-paused", - paused_undeliverable: "needs verification", - completed: "finished", - draft: "draft", -}; - -// Single source of truth for a status's color — drives BOTH the leading mark -// and the right-side text label so they always agree. emerald = live/done, -// amber = paused/needs-attention, slate = not started. -const STATUS_TONE: Record = { - active: "text-emerald-600", - completed: "text-emerald-600", - paused: "text-amber-600", - paused_no_accounts: "text-amber-600", - paused_trial_expired: "text-amber-600", - paused_guardrail: "text-rose-600", - paused_undeliverable: "text-amber-600", - draft: "text-slate-500", -}; - -function statusTone(status: string): string { - return STATUS_TONE[status] ?? STATUS_TONE.draft; -} +// fixed-width leading slot keeps each row's name aligned. The label/tone maps +// live in components/app/campaigns/status so pickers share them. +const STATUS_LABEL = CAMPAIGN_STATUS_LABEL; function CampaignStatusMark({ status }: { status: string }) { const tone = statusTone(status); diff --git a/web/src/components/app/AddContacts.tsx b/web/src/components/app/AddContacts.tsx index e415c540d..c81062fd0 100644 --- a/web/src/components/app/AddContacts.tsx +++ b/web/src/components/app/AddContacts.tsx @@ -32,6 +32,9 @@ export interface AddContact { phone: string; campaigns: string[]; categories?: string[]; + // Segments the contact is pinned into on creation (an include override), + // so a contact created from inside a segment lands in it. + segments?: string[]; custom_fields: Record; // First-touch source hint: "campaign" when added from a campaign's // Leads tab, else "manual". The server decides every other origin. diff --git a/web/src/components/app/campaigns/status.ts b/web/src/components/app/campaigns/status.ts new file mode 100644 index 000000000..466a8f6cb --- /dev/null +++ b/web/src/components/app/campaigns/status.ts @@ -0,0 +1,49 @@ +// One place every surface reads a campaign's status from, so a list row, a +// picker row and a badge never disagree (and none of them leaks the raw +// backend value like "paused_no_accounts" at the user). + +// Collapse the backend's raw campaign statuses into the buckets a list +// filters/counts by. paused_no_accounts / paused_trial_expired are auto-pause +// variants, so they belong with "paused"; anything unknown reads as draft. +export function campaignStatusBucket(s?: string): "active" | "paused" | "completed" | "draft" { + if (s === "active") return "active"; + if (s === "completed") return "completed"; + if (s && s.startsWith("paused")) return "paused"; + return "draft"; +} + +export const CAMPAIGN_STATUS_LABEL: Record = { + active: "running", + paused: "paused", + paused_no_accounts: "no accounts", + paused_trial_expired: "trial expired", + paused_guardrail: "auto-paused", + paused_undeliverable: "needs verification", + completed: "finished", + draft: "draft", +}; + +// Single source of truth for a status's color — drives BOTH the leading mark +// and the right-side text label so they always agree. emerald = live/done, +// amber = paused/needs-attention, slate = not started. +const CAMPAIGN_STATUS_TONE: Record = { + active: "text-emerald-600", + completed: "text-emerald-600", + paused: "text-amber-600", + paused_no_accounts: "text-amber-600", + paused_trial_expired: "text-amber-600", + paused_guardrail: "text-rose-600", + paused_undeliverable: "text-amber-600", + draft: "text-slate-500", +}; + +// An unmapped status still has to read as words, not an enum: underscores +// become spaces rather than surfacing "PAUSED_NO_ACCOUNTS". +export function campaignStatusLabel(status?: string): string { + if (!status) return CAMPAIGN_STATUS_LABEL.draft; + return CAMPAIGN_STATUS_LABEL[status] ?? status.replace(/_/g, " "); +} + +export function campaignStatusTone(status: string): string { + return CAMPAIGN_STATUS_TONE[status] ?? CAMPAIGN_STATUS_TONE.draft; +} diff --git a/web/src/components/app/contacts/ContactsTable.tsx b/web/src/components/app/contacts/ContactsTable.tsx index a048073f0..12fb56837 100644 --- a/web/src/components/app/contacts/ContactsTable.tsx +++ b/web/src/components/app/contacts/ContactsTable.tsx @@ -789,7 +789,7 @@ export default function ContactsTable({ /> - setNewOpen(false)} /> + setNewOpen(false)} segment={segment} /> {segment && ( buildError(err), }); onClose(); @@ -108,6 +113,11 @@ export function NewContactDialog({ open, onClose, campaign }: Props) { Contact + {segment && ( + + {segment.name} + + )} ); @@ -170,15 +178,17 @@ export default function AddSegmentToCampaignDialog({ )}
-
- + {/* Wraps instead of clipping: the hint takes its own row when the + buttons need the width, so neither is ever cut mid-word. */} +
+ Adds today's members. Contacts already in the campaign are skipped. @@ -186,7 +196,7 @@ export default function AddSegmentToCampaignDialog({ type="button" onClick={submit} disabled={busy || !picked} - className="h-7 px-2.5 rounded-md bg-sky-600 hover:bg-sky-700 text-white text-[12px] font-medium inline-flex items-center gap-1.5 transition-colors disabled:opacity-50" + className="shrink-0 whitespace-nowrap h-7 px-2.5 rounded-md bg-sky-600 hover:bg-sky-700 text-white text-[12px] font-medium inline-flex items-center gap-1.5 transition-colors disabled:opacity-50" > {busy ? : } Add leads diff --git a/web/src/components/app/segments/CampaignSegmentsDialog.tsx b/web/src/components/app/segments/CampaignSegmentsDialog.tsx index 3bfe395d4..8c65e140c 100644 --- a/web/src/components/app/segments/CampaignSegmentsDialog.tsx +++ b/web/src/components/app/segments/CampaignSegmentsDialog.tsx @@ -226,15 +226,15 @@ export default function CampaignSegmentsDialog({ )}
-