From d112657cf29d8ed6f4b6e4df1551f63bf0cd1589 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sun, 4 Oct 2026 09:10:37 +0200 Subject: [PATCH] feat: wire suppression, subscription and hold dependencies into the consumer's Salesforce service, give Salesforce token refreshes their own single-flight keyspace, release the outbox lease in memory after each terminal write so writeback notes land and only on rows with a logged Task, drop stale Salesforce import previews, keep keys from the import row menu from opening the edit dialog, and document the Salesforce callback URL and OAuth variables --- cmd/consumer/main.go | 3 +++ docs/content/docs/development/configuration.mdx | 1 + docs/content/docs/guides/salesforce.mdx | 2 +- internal/app/integration/service.go | 3 ++- internal/app/salesforce/drain.go | 7 ++++++- .../integrations/salesforce/_components/ImportDialog.tsx | 8 ++++++-- .../app/integrations/salesforce/_components/ImportTab.tsx | 7 ++++++- 7 files changed, 25 insertions(+), 6 deletions(-) diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index 17bfbaefe..1aaa5b232 100644 --- a/cmd/consumer/main.go +++ b/cmd/consumer/main.go @@ -248,6 +248,9 @@ func main() { Repo: repository.NewSalesforceRepository(primaryDB.Pool), Integrations: integrationServiceC, Cipher: cipherService, + Holds: campaignProgressRepo, + Suppression: advancedRepo, + Subscription: contactRepo, }) integrationServiceC.SetSalesforce(salesforceC) webhookService.WireRecordSink(salesforceC.Recorder().Record) diff --git a/docs/content/docs/development/configuration.mdx b/docs/content/docs/development/configuration.mdx index 0eded7db5..0dfc9d2fc 100644 --- a/docs/content/docs/development/configuration.mdx +++ b/docs/content/docs/development/configuration.mdx @@ -508,6 +508,7 @@ Application permissions reach every mailbox in a consenting organization. We rec |---|---|---| | `_OAUTH_CLIENT_ID`, `_OAUTH_CLIENT_SECRET` | OAuth clients for the CRM and messaging integrations | unset | | `HUBSPOT_OAUTH_CLIENT_ID`, `HUBSPOT_OAUTH_CLIENT_SECRET` | Your HubSpot app. Needed on the backend and on every consumer: the consumer runs the [HubSpot mode](/guides/hubspot/) pull and sync queue and refreshes the token as it goes. The secret also verifies HubSpot's webhook signatures | unset | +| `SALESFORCE_OAUTH_CLIENT_ID`, `SALESFORCE_OAUTH_CLIENT_SECRET` | Your Salesforce app (see [Salesforce self-hosting](/guides/salesforce/#self-hosting)). Needed on the backend, which runs the sync, and on every consumer, which runs Salesforce automation actions and refreshes the token for them | unset | | `INTEGRATIONS_OAUTH_REDIRECT_URL` | Shared redirect URI for those flows | `BACKEND_PUBLIC_URL` (else `API_PUBLIC_URL`) plus `/integrations/oauth/callback` | | `SLACK_OAUTH_CLIENT_ID`, `SLACK_OAUTH_CLIENT_SECRET` | The instance's own [Slack app](/development/slack-app/). Unset, the dashboard reports Slack as not set up and no workspace can connect it | unset | | `SLACK_SIGNING_SECRET` | Verifies every request Slack sends to the events, interactivity and slash command URLs. Without it Slack still posts notifications and inbox conversations, but the assistant, buttons and `/warmbly` are off, and those URLs answer `503` `slack_not_configured` | unset | diff --git a/docs/content/docs/guides/salesforce.mdx b/docs/content/docs/guides/salesforce.mdx index 706b2baea..1a7c51bda 100644 --- a/docs/content/docs/guides/salesforce.mdx +++ b/docs/content/docs/guides/salesforce.mdx @@ -147,7 +147,7 @@ A self-hosted instance needs its own Salesforce app. Which kind depends on how m Whichever you create, configure it with: -- callback URL `/integrations/oauth/callback`, or your `INTEGRATIONS_OAUTH_REDIRECT_URL`. It must match exactly, including the scheme and no trailing slash; +- callback URL: the value of `INTEGRATIONS_OAUTH_REDIRECT_URL`, or when that is unset, `BACKEND_PUBLIC_URL` (else `API_PUBLIC_URL`) plus `/integrations/oauth/callback`. It must match exactly, including the scheme and no trailing slash; - OAuth scopes **Manage user data via APIs (api)**, **Perform requests at any time (refresh_token, offline_access)** and **Access the identity URL service (id, profile, email, address, phone)**; - **Require PKCE** on, and the **Web Server Flow** enabled with its client secret required; - refresh tokens **valid until revoked**. diff --git a/internal/app/integration/service.go b/internal/app/integration/service.go index 3f812adde..72df976ca 100644 --- a/internal/app/integration/service.go +++ b/internal/app/integration/service.go @@ -1262,7 +1262,8 @@ func (s *service) ProviderAccess(ctx context.Context, orgID, connID uuid.UUID, f } // Concurrent callers share one refresh: with refresh-token rotation, two // parallel refreshes would leave one holding a dead token. - key := connID.String() + // Its own keyspace: AccessToken shares the group with a different result type. + key := "provider:" + connID.String() if force { key += ":force" } diff --git a/internal/app/salesforce/drain.go b/internal/app/salesforce/drain.go index fe64d4b19..8aad09042 100644 --- a/internal/app/salesforce/drain.go +++ b/internal/app/salesforce/drain.go @@ -627,7 +627,9 @@ func (s *Service) writeback(ctx context.Context, c *conn, work []*pending, links if e := errs[p.contact.ID]; e != nil { if p.done { // The Task landed; the record update did not. Say so on the row. - s.annotate(ctx, p, "Task logged, but updating the record failed: "+e.Error()) + if p.a.TaskID != "" { + s.annotate(ctx, p, "Task logged, but updating the record failed: "+describeErr(e)) + } continue } s.fail(ctx, p, e) @@ -649,6 +651,8 @@ func (s *Service) finish(ctx context.Context, p *pending, status, taskID, detail } p.a.Attempts++ _ = s.Repo.FinishActivity(ctx, &p.a) + // FinishActivity released the lease; a later note on this row targets it unleased. + p.a.LeaseID = nil } func (s *Service) annotate(ctx context.Context, p *pending, detail string) { @@ -687,6 +691,7 @@ func (s *Service) fail(ctx context.Context, p *pending, err error) { p.a.Status = models.SalesforceActivityFailed } _ = s.Repo.FinishActivity(ctx, &p.a) + p.a.LeaseID = nil } func backoff(attempt int) time.Duration { diff --git a/web/src/app/app/integrations/salesforce/_components/ImportDialog.tsx b/web/src/app/app/integrations/salesforce/_components/ImportDialog.tsx index 2402e2813..d98142aba 100644 --- a/web/src/app/app/integrations/salesforce/_components/ImportDialog.tsx +++ b/web/src/app/app/integrations/salesforce/_components/ImportDialog.tsx @@ -109,13 +109,17 @@ export default function ImportDialog({ connectionId, onClose }: { connectionId: return () => document.removeEventListener("keydown", onKey); }, [requestClose]); + // The selection the newest preview request was for; an older answer is dropped. + const latestPreview = React.useRef(""); async function loadPreview() { const parts = kindParts(kind); - setPreviewFor(selectionKey); + const key = selectionKey; + latestPreview.current = key; + setPreviewFor(key); setPreview(null); try { const p = await previewM.mutateAsync({ connectionId, ...parts, source_id: sourceId }); - setPreview(p); + if (latestPreview.current === key) setPreview(p); } catch { // Rendered in the step from previewM.error. } diff --git a/web/src/app/app/integrations/salesforce/_components/ImportTab.tsx b/web/src/app/app/integrations/salesforce/_components/ImportTab.tsx index 455bf9588..3ff872cbe 100644 --- a/web/src/app/app/integrations/salesforce/_components/ImportTab.tsx +++ b/web/src/app/app/integrations/salesforce/_components/ImportTab.tsx @@ -174,7 +174,12 @@ function SourceRow({ tabIndex={0} onClick={onEdit} onKeyDown={(e) => { - if (e.key === "Enter") onEdit(); + // Only the row itself: keys from its menu bubble here through the portal. + if (e.target !== e.currentTarget) return; + if (e.key === "Enter" || e.key === " ") { + e.preventDefault(); + onEdit(); + } }} className="px-4 py-3 flex items-start gap-3 hover:bg-slate-50/60 transition-colors cursor-pointer outline-none focus-visible:bg-slate-50" >