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 2ef79e585..fc63997fc 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/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/docs/content/docs/api/reference/campaigns.mdx b/docs/content/docs/api/reference/campaigns.mdx
index f4a4dbe1f..3148172fd 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. 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 |
+| --- | --- | --- | --- |
+| `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..2f7aea2f0 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, and the override is written in the same transaction as the contact, so a success response always means the membership exists. |
| `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`. |
@@ -366,6 +367,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 +1030,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 +1070,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..b0a22e916 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. 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. Neither offers to delete the contact itself: inside a campaign the destructive action is leaving the campaign, and deleting a contact from the workspace is done from the contacts list.
### Lead statuses
diff --git a/docs/content/docs/guides/contacts-crm.mdx b/docs/content/docs/guides/contacts-crm.mdx
index a5f702c7d..1ade6e79f 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.
@@ -69,6 +71,12 @@ See [Personalization & expressions](/guides/expressions/) for the full templatin
A filter bar sits above the contact list. **Category**, **Segment**, **Status** and **Campaign** are always there; open one, tick values, and the list updates immediately with the matching count next to the bar. **Add filter** adds a custom-field condition (field, contains/is/starts with/ends with, value), a date-added or last-updated range, a number-of-campaigns range, the address verification verdict and, on a campaign's Leads tab, lead status and engagement. Each active filter is a pill you can reopen to change or remove with its cross; **Clear** drops them all, and **Save as segment** turns the current set into a [segment](/guides/segments/). Free-text search and sort stay in the toolbar.
+## Editing many contacts at once
+
+Tick rows in any contact list, a segment's members or a campaign's leads, and the selection bar's **Edit** opens a bulk panel. It can add or remove campaigns, add or remove categories, force a subscription state, and queue custom-field operations (add, edit, delete, rename) that run on every selected contact.
+
+Nothing is applied while you build it up. The header names the list the selection came from, and a **Will apply** strip above the buttons lists every queued change as a chip you can take back one at a time before pressing Apply. A field operation missing its key, or its value where one is needed, is marked and skipped rather than sent half-finished.
+
## Categories
Colored labels that group and filter contacts (`Warm lead`, `Conference 2026`, `Enterprise`), behaving like tags. The same picker appears in bulk edit, a contact's Details tab, the new-contact dialog, the filters, and the import and sync wizards.
diff --git a/docs/content/docs/guides/segments.mdx b/docs/content/docs/guides/segments.mdx
index ba84cdbcc..a79b614f0 100644
--- a/docs/content/docs/guides/segments.mdx
+++ b/docs/content/docs/guides/segments.mdx
@@ -42,19 +42,38 @@ 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.
+**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
- **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. 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:
+
+- **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..a565527f0 100644
--- a/internal/api/handler/segment.go
+++ b/internal/api/handler/segment.go
@@ -207,6 +207,62 @@ 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
+ }
+ // 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)
+ 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 cf3522c60..61dd357f2 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..053967f34 100644
--- a/internal/app/contact/handler.go
+++ b/internal/app/contact/handler.go
@@ -56,6 +56,14 @@ func (s *contactService) Add(ctx context.Context, userID string, orgID uuid.UUID
return nil, xerr
}
+ // Checked before the write so an unknown segment is a 400 rather than a
+ // silently-skipped link. The override itself is written inside the
+ // repository's transaction, so a contact is never created without the
+ // segment membership the caller asked for.
+ if xerr := s.checkSegmentTargets(ctx, orgID, contacts); xerr != nil {
+ return nil, xerr
+ }
+
created, xerr := s.contactRepository.Add(ctx, userID, orgID, contacts)
if xerr != nil {
return nil, xerr
@@ -67,9 +75,26 @@ 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
}
+// checkSegmentTargets rejects a create whose segments do not exist in the
+// organization. Existence is checked once for the whole batch, so N contacts
+// naming the same segment cost one lookup; the repository writes the overrides
+// itself, inside the same transaction as the contacts.
+func (s *contactService) checkSegmentTargets(ctx context.Context, orgID uuid.UUID, contacts []models.AddContact) *errx.Error {
+ var raw []string
+ for i := range contacts {
+ raw = append(raw, contacts[i].Segments...)
+ }
+ if len(raw) == 0 {
+ return nil
+ }
+ _, xerr := s.parseSegmentIDs(ctx, orgID, raw)
+ return xerr
+}
+
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 {
@@ -135,6 +160,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 +172,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..f2a9f544e 100644
--- a/internal/app/orgtransfer/spec.go
+++ b/internal/app/orgtransfer/spec.go
@@ -346,6 +346,17 @@ 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_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 17275c722..d8ed18ad4 100644
--- a/internal/app/segment/service.go
+++ b/internal/app/segment/service.go
@@ -1,13 +1,18 @@
// 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"
+ "sync"
+ "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 +23,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 +43,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 +65,31 @@ type CustomFieldLister interface {
}
type service struct {
- repo repository.SegmentRepository
- fields CustomFieldLister
- waker CampaignWaker
+ repo repository.SegmentRepository
+ fields CustomFieldLister
+ waker CampaignWaker
+ starter CampaignStarter
+ // orgSync coalesces org-wide enrolment passes, one entry per org that is
+ // currently syncing. Guarded by syncMu, which owns every transition so an
+ // entry is only dropped when nothing is running or queued.
+ syncMu sync.Mutex
+ orgSync map[uuid.UUID]*orgSyncState
+}
+
+// orgSyncState is a running pass plus a "do it once more" flag. A write that
+// lands mid-pass cannot be served by that pass (it read the old membership),
+// so it asks for a follow-up instead of being dropped.
+type orgSyncState struct {
+ running bool
+ again bool
}
func NewService(repo repository.SegmentRepository, fields CustomFieldLister) Service {
- return &service{repo: repo, fields: fields}
+ return &service{repo: repo, fields: fields, orgSync: map[uuid.UUID]*orgSyncState{}}
}
-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 +205,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 +225,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 +290,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 +327,190 @@ 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. 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) {
+ 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) {
+ // One pass per org at a time, so a burst of contact writes cannot stack
+ // org-wide enrolment scans. Requests that arrive mid-pass are coalesced
+ // into a single follow-up rather than dropped: an import writes its
+ // segment membership after the contact rows, so the pass already running
+ // read the old membership and would leave those contacts to the sweep.
+ s.syncMu.Lock()
+ st := s.orgSync[orgID]
+ if st == nil {
+ st = &orgSyncState{}
+ s.orgSync[orgID] = st
+ }
+ if st.running {
+ st.again = true
+ s.syncMu.Unlock()
+ return
+ }
+ st.running = true
+ s.syncMu.Unlock()
+
+ bg := context.WithoutCancel(ctx)
+ go func() {
+ for {
+ s.runOrgSyncPass(bg, orgID)
+ s.syncMu.Lock()
+ if st.again {
+ st.again = false
+ s.syncMu.Unlock()
+ continue
+ }
+ delete(s.orgSync, orgID)
+ s.syncMu.Unlock()
+ return
+ }
+ }()
+}
+
+// runOrgSyncPass enrols every linked campaign in one organization once.
+func (s *service) runOrgSyncPass(ctx context.Context, orgID uuid.UUID) {
+ rctx, cancel := context.WithTimeout(ctx, 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) {
+ 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) {
+ 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 {
+ 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")
+ }
+}
+
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/app/segment/service_test.go b/internal/app/segment/service_test.go
new file mode 100644
index 000000000..45d816989
--- /dev/null
+++ b/internal/app/segment/service_test.go
@@ -0,0 +1,125 @@
+package segment
+
+import (
+ "context"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/google/uuid"
+
+ "github.com/warmbly/warmbly/internal/errx"
+ "github.com/warmbly/warmbly/internal/models"
+ "github.com/warmbly/warmbly/internal/repository"
+)
+
+// syncRepo is a SegmentRepository that only answers LinkedCampaigns, blocking
+// there until released so a second request provably arrives mid-pass. Every
+// other method is left to the embedded nil interface: this test must not reach
+// them, and a panic is a clearer failure than a silent zero value.
+type syncRepo struct {
+ repository.SegmentRepository
+ entered chan struct{}
+ release chan struct{}
+
+ mu sync.Mutex
+ calls int
+}
+
+func (r *syncRepo) LinkedCampaigns(context.Context, *uuid.UUID) ([]models.LinkedCampaign, *errx.Error) {
+ r.mu.Lock()
+ r.calls++
+ r.mu.Unlock()
+ r.entered <- struct{}{}
+ <-r.release
+ return nil, nil
+}
+
+func (r *syncRepo) callCount() int {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ return r.calls
+}
+
+func waitFor(t *testing.T, c chan struct{}, what string) {
+ t.Helper()
+ select {
+ case <-c:
+ case <-time.After(2 * time.Second):
+ t.Fatalf("timed out waiting for %s", what)
+ }
+}
+
+// A sync requested while a pass is running must run again afterwards. The
+// import path depends on it: ImportCommit's chunked Add starts a pass, then
+// writes segment membership, then asks for a sync. Dropping that request
+// leaves freshly pinned contacts out of their linked campaigns until the
+// periodic sweep.
+func TestSyncOrgLinkedCampaignsCoalescesMidPassRequest(t *testing.T) {
+ repo := &syncRepo{entered: make(chan struct{}, 4), release: make(chan struct{})}
+ svc := NewService(repo, nil).(*service)
+ org := uuid.New()
+ ctx := context.Background()
+
+ svc.SyncOrgLinkedCampaigns(ctx, org)
+ waitFor(t, repo.entered, "the first pass to start")
+
+ // Two requests land while the first pass is blocked: both are folded into
+ // one follow-up, not stacked and not lost.
+ svc.SyncOrgLinkedCampaigns(ctx, org)
+ svc.SyncOrgLinkedCampaigns(ctx, org)
+ if got := repo.callCount(); got != 1 {
+ t.Fatalf("passes started while one was running = %d, want 1", got)
+ }
+
+ repo.release <- struct{}{}
+ waitFor(t, repo.entered, "the coalesced follow-up pass")
+ repo.release <- struct{}{}
+
+ // The follow-up drains the flag, so nothing runs a third time.
+ deadline := time.After(300 * time.Millisecond)
+ for {
+ select {
+ case <-repo.entered:
+ t.Fatalf("a third pass ran; the follow-up flag was not cleared")
+ case <-deadline:
+ if got := repo.callCount(); got != 2 {
+ t.Fatalf("total passes = %d, want 2", got)
+ }
+ svc.syncMu.Lock()
+ _, still := svc.orgSync[org]
+ svc.syncMu.Unlock()
+ if still {
+ t.Errorf("org sync state was not released once idle")
+ }
+ return
+ }
+ }
+}
+
+// Back-to-back requests with no overlap each get their own pass.
+func TestSyncOrgLinkedCampaignsRunsAgainAfterIdle(t *testing.T) {
+ repo := &syncRepo{entered: make(chan struct{}, 4), release: make(chan struct{})}
+ svc := NewService(repo, nil).(*service)
+ org := uuid.New()
+ ctx := context.Background()
+
+ for i := 0; i < 2; i++ {
+ svc.SyncOrgLinkedCampaigns(ctx, org)
+ waitFor(t, repo.entered, "a pass to start")
+ repo.release <- struct{}{}
+ // Let the goroutine retire the state before asking again.
+ for j := 0; j < 100; j++ {
+ svc.syncMu.Lock()
+ _, running := svc.orgSync[org]
+ svc.syncMu.Unlock()
+ if !running {
+ break
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ }
+ if got := repo.callCount(); got != 2 {
+ t.Fatalf("passes = %d, want 2", got)
+ }
+}
diff --git a/internal/infrastructure/db/migrations/000116_campaign_segments.down.sql b/internal/infrastructure/db/migrations/000116_campaign_segments.down.sql
new file mode 100644
index 000000000..568be908e
--- /dev/null
+++ b/internal/infrastructure/db/migrations/000116_campaign_segments.down.sql
@@ -0,0 +1 @@
+DROP TABLE IF EXISTS campaign_segments;
diff --git a/internal/infrastructure/db/migrations/000116_campaign_segments.up.sql b/internal/infrastructure/db/migrations/000116_campaign_segments.up.sql
new file mode 100644
index 000000000..f84573511
--- /dev/null
+++ b/internal/infrastructure/db/migrations/000116_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/infrastructure/db/migrations/000117_campaign_lead_removals.down.sql b/internal/infrastructure/db/migrations/000117_campaign_lead_removals.down.sql
new file mode 100644
index 000000000..07c4647d9
--- /dev/null
+++ b/internal/infrastructure/db/migrations/000117_campaign_lead_removals.down.sql
@@ -0,0 +1 @@
+DROP TABLE IF EXISTS campaign_lead_removals;
diff --git a/internal/infrastructure/db/migrations/000117_campaign_lead_removals.up.sql b/internal/infrastructure/db/migrations/000117_campaign_lead_removals.up.sql
new file mode 100644
index 000000000..74232fbe9
--- /dev/null
+++ b/internal/infrastructure/db/migrations/000117_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/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"},
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/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_contact.go b/internal/repository/pg_contact.go
index 0505b4d38..fd0068d3a 100644
--- a/internal/repository/pg_contact.go
+++ b/internal/repository/pg_contact.go
@@ -166,6 +166,7 @@ func (r *contactRepository) Add(ctx context.Context, userID string, orgID uuid.U
normalized := make([]models.AddContact, 0, len(contacts))
campaignIDs := make([][]uuid.UUID, 0, len(contacts))
categoryIDs := make([][]uuid.UUID, 0, len(contacts))
+ segmentIDs := make([][]uuid.UUID, 0, len(contacts))
for _, lead := range contacts {
lead.Email = strings.TrimSpace(lead.Email)
if !email.IsValid(lead.Email) {
@@ -236,6 +237,27 @@ func (r *contactRepository) Add(ctx context.Context, userID string, orgID uuid.U
cats = append(cats, cid)
}
+ // Parse + dedupe segment IDs. Same rules again; these become manual
+ // include overrides written in this transaction, so a contact created
+ // inside a segment cannot come back without it.
+ segSet := make(map[uuid.UUID]struct{}, len(lead.Segments))
+ segs := make([]uuid.UUID, 0, len(lead.Segments))
+ for _, raw := range lead.Segments {
+ raw = strings.TrimSpace(raw)
+ if raw == "" {
+ continue
+ }
+ sid, serr := uuid.Parse(raw)
+ if serr != nil {
+ return nil, errx.ErrUuid
+ }
+ if _, dup := segSet[sid]; dup {
+ continue
+ }
+ segSet[sid] = struct{}{}
+ segs = append(segs, sid)
+ }
+
// First-touch attribution. Every creation path stamps one; an empty
// value is recorded honestly rather than guessed.
if lead.Source == "" {
@@ -256,6 +278,7 @@ func (r *contactRepository) Add(ctx context.Context, userID string, orgID uuid.U
normalized = append(normalized, lead)
campaignIDs = append(campaignIDs, cids)
categoryIDs = append(categoryIDs, cats)
+ segmentIDs = append(segmentIDs, segs)
}
tx, err := r.DB.Begin(ctx)
@@ -481,6 +504,34 @@ func (r *contactRepository) Add(ctx context.Context, userID string, orgID uuid.U
ncontacts[i].Categories = linked
}
+ // Pin into segments, in the same transaction as the contact itself: a
+ // failed override write must not answer with a contact that quietly never
+ // joined the segment it was created in (issue #285). Scoped to the org, so
+ // a foreign segment id links nothing.
+ for i, segs := range segmentIDs {
+ if len(segs) == 0 {
+ continue
+ }
+ tag, err := tx.Exec(ctx, `
+ INSERT INTO segment_members (segment_id, contact_id, mode)
+ SELECT s.id, $1, 'include'
+ FROM segments s
+ WHERE s.id = ANY($2) AND s.organization_id = $3
+ ON CONFLICT (segment_id, contact_id) DO UPDATE SET mode = EXCLUDED.mode, created_at = now()
+ `, ncontacts[i].ID, segs, orgID)
+ if err != nil {
+ db.CaptureError(err, "", nil, "segment_members insert")
+ return nil, errx.InternalError()
+ }
+ // One row per requested segment, since `segs` is deduplicated and a
+ // conflicting row still counts as affected. Fewer means a segment was
+ // deleted between the service's check and this statement, so the
+ // create is rolled back rather than answered without the membership.
+ if int(tag.RowsAffected()) != len(segs) {
+ return nil, errx.New(errx.BadRequest, "a selected segment does not exist")
+ }
+ }
+
if err := logContactCreated(ctx, tx, orgID, actor, createdIDs); err != nil {
db.CaptureError(err, "", nil, "contact_created activity")
return nil, errx.InternalError()
@@ -1828,14 +1879,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 +1913,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 +2177,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 +2189,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 5abbb7331..ba3db5b35 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, false)
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,287 @@ 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.
+//
+// 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)%s
+ ON CONFLICT DO NOTHING
+ RETURNING contact_id, campaign_id`, len(args), clause, guard)
+ 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)
+ }
+ // 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 {
+ 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()
+ // 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 {
+ 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, true)
+ 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)
+ }
+ if err := rows.Err(); err != nil {
+ db.CaptureError(err, "linked campaigns", nil, "rows")
+ return nil, errx.InternalError()
+ }
+ 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)
+ }
+ if err := rows.Err(); err != nil {
+ db.CaptureError(err, "campaigns using segment", nil, "rows")
+ return nil, errx.InternalError()
+ }
+ 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..a68a815f0 100644
--- a/internal/repository/segment_live_test.go
+++ b/internal/repository/segment_live_test.go
@@ -241,3 +241,164 @@ 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)
+ }
+
+ // 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)
+ 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)
+ }
+}
+
+// Issue #285: a contact created inside a segment joins it, and the include
+// override is written in the same transaction as the contact. A segment that
+// vanishes between the service's existence check and this insert must roll the
+// whole create back rather than answer with a membership that never happened.
+func TestLiveContactAddPinsSegments(t *testing.T) {
+ f, repo := newSegmentFixture(t)
+ ctx := context.Background()
+ handle, pool := liveContactDB(t)
+ contacts := NewContactRepostory(handle)
+
+ acme, xerr := repo.Create(ctx, f.org, &f.owner, &models.Segment{
+ Name: "Pin target", Color: "#0284c7", Match: models.SegmentMatchAll,
+ Conditions: []models.SegmentCondition{{Field: "company", Operator: "equals", Value: "acme"}},
+ })
+ if xerr != nil {
+ t.Fatalf("create segment: %v", xerr)
+ }
+
+ // A company the conditions do not match, so membership can only come from
+ // the pin. The second contact names no segment and must stay out.
+ tag := uuid.New().String()[:6]
+ made, xerr := contacts.Add(ctx, f.owner.String(), f.org, []models.AddContact{
+ {Email: "pinned-" + tag + "@initech.test", FirstName: "Pinned", Company: "initech", Segments: []string{acme.ID.String(), acme.ID.String()}},
+ {Email: "loose-" + tag + "@initech.test", FirstName: "Loose", Company: "initech"},
+ })
+ if xerr != nil || len(made) != 2 {
+ t.Fatalf("add: %+v, %v", made, xerr)
+ }
+ modes, xerr := repo.MemberModes(ctx, acme.ID, []uuid.UUID{made[0].ID, made[1].ID})
+ if xerr != nil {
+ t.Fatalf("modes: %v", xerr)
+ }
+ if modes[made[0].ID] != models.SegmentMemberInclude {
+ t.Errorf("pinned contact mode = %q, want include", modes[made[0].ID])
+ }
+ if _, ok := modes[made[1].ID]; ok {
+ t.Errorf("contact that named no segment was pinned")
+ }
+
+ // A segment id that is not in the organization (the state the race leaves
+ // behind) fails the create instead of dropping the membership silently.
+ gone := uuid.New().String()
+ email := "vanished-" + tag + "@initech.test"
+ if _, xerr := contacts.Add(ctx, f.owner.String(), f.org, []models.AddContact{
+ {Email: email, FirstName: "Vanished", Company: "initech", Segments: []string{gone}},
+ }); xerr == nil {
+ t.Fatalf("create with a vanished segment succeeded")
+ }
+ var stranded int
+ if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM contacts WHERE organization_id = $1 AND email = $2`, f.org, email).Scan(&stranded); err != nil {
+ t.Fatalf("count: %v", err)
+ }
+ if stranded != 0 {
+ t.Errorf("rolled-back create left %d contacts behind", stranded)
+ }
+}
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/CategoryPicker.tsx b/web/src/components/app/contacts/CategoryPicker.tsx
index 7d9998f06..db44f6b2a 100644
--- a/web/src/components/app/contacts/CategoryPicker.tsx
+++ b/web/src/components/app/contacts/CategoryPicker.tsx
@@ -136,6 +136,7 @@ export default function CategoryPicker({
{open && (
0;
+}
+
+const plural = (n: number, one: string, many: string) => `${n} ${n === 1 ? one : many}`;
+
export default function ContactsEditBulk({
active,
setActive,
selected,
+ scope,
}: {
active: boolean;
setActive: React.Dispatch>;
selected: string[];
+ // Where the selection was made, shown in the header so a bulk write is
+ // never applied without knowing which list it came from.
+ scope?: { kind: "segment" | "campaign"; name: string };
}) {
- const [campaignsAdd, setCampaignsAdd] = React.useState([]);
- const [campaignsRemove, setCampaignsRemove] = React.useState([]);
+ const [campaignsAdd, setCampaignsAdd] = React.useState([]);
+ const [campaignsRemove, setCampaignsRemove] = React.useState([]);
const [categoriesAdd, setCategoriesAdd] = React.useState([]);
const [categoriesRemove, setCategoriesRemove] = React.useState([]);
const [fields, setFields] = React.useState([]);
@@ -65,63 +81,87 @@ export default function ContactsEditBulk({
const update = useUpdateContactsBulk();
const confirm = useConfirm();
- const dirty =
- campaignsAdd.length > 0 ||
- campaignsRemove.length > 0 ||
- categoriesAdd.length > 0 ||
- categoriesRemove.length > 0 ||
- fields.length > 0 ||
- subscribeMode !== "unchanged";
-
- // Close, guarding unsaved edits behind the in-app confirm (never window.confirm).
- const requestClose = React.useCallback(() => {
- if (dirty) confirm.show("Discard bulk changes?", () => setActive(false));
- else setActive(false);
- }, [dirty, confirm, setActive]);
-
- function reset() {
+ const reset = React.useCallback(() => {
setCampaignsAdd([]);
setCampaignsRemove([]);
setCategoriesAdd([]);
setCategoriesRemove([]);
setFields([]);
setSubscribeMode("unchanged");
- }
+ }, []);
+
+ const readyFields = React.useMemo(() => fields.filter(isComplete), [fields]);
+ const incomplete = fields.length - readyFields.length;
+
+ // What the Apply button will actually do, in the order it reads. Each entry
+ // can be taken back without hunting for the control that set it.
+ const changes = React.useMemo(() => {
+ const out: { key: string; label: string; clear: () => void }[] = [];
+ if (campaignsAdd.length > 0)
+ out.push({ key: "ca", label: `Add to ${plural(campaignsAdd.length, "campaign", "campaigns")}`, clear: () => setCampaignsAdd([]) });
+ if (campaignsRemove.length > 0)
+ out.push({ key: "cr", label: `Remove from ${plural(campaignsRemove.length, "campaign", "campaigns")}`, clear: () => setCampaignsRemove([]) });
+ if (categoriesAdd.length > 0)
+ out.push({ key: "ka", label: `Add ${plural(categoriesAdd.length, "category", "categories")}`, clear: () => setCategoriesAdd([]) });
+ if (categoriesRemove.length > 0)
+ out.push({ key: "kr", label: `Remove ${plural(categoriesRemove.length, "category", "categories")}`, clear: () => setCategoriesRemove([]) });
+ if (subscribeMode !== "unchanged")
+ out.push({ key: "s", label: subscribeMode === "subscribe" ? "Subscribe" : "Unsubscribe", clear: () => setSubscribeMode("unchanged") });
+ if (readyFields.length > 0)
+ out.push({ key: "f", label: `${plural(readyFields.length, "field change", "field changes")}`, clear: () => setFields([]) });
+ return out;
+ }, [campaignsAdd, campaignsRemove, categoriesAdd, categoriesRemove, subscribeMode, readyFields]);
+
+ // Anything the user touched guards the close; only complete changes can be
+ // applied, so the two are deliberately not the same question.
+ const touched = changes.length > 0 || fields.length > 0;
+ const applicable = changes.length > 0;
+
+ const requestClose = React.useCallback(() => {
+ if (touched) confirm.show("Discard bulk changes?", async () => setActive(false));
+ else setActive(false);
+ }, [touched, confirm, setActive]);
async function submit() {
- if (!dirty) return;
- const data = {
- contacts: selected,
- add_campaigns: campaignsAdd.map((c) => c.id),
- remove_campaigns: campaignsRemove.map((c) => c.id),
- add_categories: categoriesAdd,
- remove_categories: categoriesRemove,
- fields,
- subscribe:
- subscribeMode === "subscribe" ? true : subscribeMode === "unsubscribe" ? false : undefined,
- };
+ if (!applicable) return;
try {
- await toast.promise(update.mutateAsync(data), {
- loading: `Updating ${selected.length} contacts…`,
- success: `Updated ${selected.length} contacts`,
- error: (err: AppError) => buildError(err),
- });
+ await toast.promise(
+ update.mutateAsync({
+ contacts: selected,
+ add_campaigns: campaignsAdd,
+ remove_campaigns: campaignsRemove,
+ add_categories: categoriesAdd,
+ remove_categories: categoriesRemove,
+ fields: readyFields,
+ subscribe: subscribeMode === "subscribe" ? true : subscribeMode === "unsubscribe" ? false : undefined,
+ }),
+ {
+ loading: `Updating ${plural(selected.length, "contact", "contacts")}…`,
+ success: `Updated ${plural(selected.length, "contact", "contacts")}`,
+ error: (err: AppError) => buildError(err),
+ },
+ );
reset();
setActive(false);
} catch {
- /* surfaced */
+ /* surfaced by toast.promise */
}
}
React.useEffect(() => {
if (!active) reset();
- }, [active]);
+ }, [active, reset]);
+ // Escape closes the innermost layer only: a picker dropdown or the confirm
+ // owns the key while it is on screen.
React.useEffect(() => {
+ if (!active) return;
function onKey(e: KeyboardEvent) {
- if (e.key === "Escape") requestClose();
+ if (e.key !== "Escape") return;
+ if (document.querySelector("[data-floating], [role='alertdialog']")) return;
+ requestClose();
}
- if (active) window.addEventListener("keydown", onKey);
+ window.addEventListener("keydown", onKey);
return () => window.removeEventListener("keydown", onKey);
}, [requestClose, active]);
@@ -139,6 +179,9 @@ export default function ContactsEditBulk({
>
-
- Bulk edit
+ Bulk edit
+
+
+ {plural(selected.length, "contact", "contacts")}
-
-
- {selected.length} {selected.length === 1 ? "contact" : "contacts"}
-
- {dirty && (
-
- Pending
+ {scope && (
+
+
+ in {scope.name}
+
)}
-