From f9a9163891b2f0baffd57eae49a108e2ef3d24d7 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sun, 11 Oct 2026 12:59:05 +0200 Subject: [PATCH] feat: record Apollo import outcomes with an uncancellable context, bound recurring runs and release interrupted ones so they rerun and sync on schedule, send the saved Clay auth token only to the connection's own table webhook, use typed routes for the Apollo, Clay and Salesforce page links, and document Clay HTTP API column retries --- docs/content/docs/guides/clay.mdx | 4 +- internal/api/handler/clay.go | 2 +- internal/app/apollo/service.go | 18 ++++- internal/app/integration/clay.go | 6 +- internal/repository/pg_apollo.go | 17 ++++- .../_components/ConnectionDetail.tsx | 76 +++++++++---------- web/src/app/app/integrations/page.tsx | 6 +- .../app/automations/AutomationEditors.tsx | 2 +- 8 files changed, 78 insertions(+), 53 deletions(-) diff --git a/docs/content/docs/guides/clay.mdx b/docs/content/docs/guides/clay.mdx index a63da7b09..227d2f054 100644 --- a/docs/content/docs/guides/clay.mdx +++ b/docs/content/docs/guides/clay.mdx @@ -94,7 +94,7 @@ Clay can map the response into columns. A single-row request answers with that r `status` is `created`, `updated`, `skipped` (for example, already a contact with **Leave it alone** chosen, or suppressed) or `failed`. A skipped or failed row carries a `reason`. A request for many rows has the same counts and one entry per row in `results`. -A URL Warmbly does not know answers `404`, a disconnected connection `403`, and more than 600 requests a minute on one connection `429` with `Retry-After`. Clay retries those on its own. +A URL Warmbly does not know answers `404` and a disconnected connection `403`. Neither goes away on a retry: fix the URL in the column or reconnect Clay, then rerun the rows. More than 600 requests a minute on one connection answer `429` with `Retry-After`, and a Warmbly error answers `500`. Those are worth retrying, so turn on retries in the HTTP API column's settings and include `429` and `500` in the statuses it retries. ### What rows become @@ -134,7 +134,7 @@ Choose the events to send, save, and use **Send a test row** to check that Clay The events are: prospect replies, email sent, email opened, link clicked, email bounces, contact unsubscribes, meeting booked and meeting canceled. Each one adds a row with the event, the contact's email and name, the campaign, and the event's details as flat columns. -To send to more than one Clay table, use an [automation](/guides/automations/) with the **Send a row to Clay** action. It sends to the connection's table unless the step names another table webhook. +To send to more than one Clay table, use an [automation](/guides/automations/) with the **Send a row to Clay** action. It sends to the connection's table unless the step names another table webhook. The connection's auth token is only sent to its own table, so a table named on a step needs a webhook source without a token. **Stop sending** removes the webhook URL, the token and the chosen events. diff --git a/internal/api/handler/clay.go b/internal/api/handler/clay.go index 0abe94b76..cba8de206 100644 --- a/internal/api/handler/clay.go +++ b/internal/api/handler/clay.go @@ -113,7 +113,7 @@ func (h *Handler) InboundClay(c *gin.Context) { c.JSON(int(xe.Code), gin.H{"status": "failed", "error": xe.Message}) default: log.Error().Err(err).Msg("clay inbound failed") - c.JSON(http.StatusInternalServerError, gin.H{"status": "failed", "error": "Warmbly could not save these rows. Clay will retry."}) + c.JSON(http.StatusInternalServerError, gin.H{"status": "failed", "error": "Warmbly could not save these rows. Run them again."}) } return } diff --git a/internal/app/apollo/service.go b/internal/app/apollo/service.go index 5587bbb84..36ffdba3d 100644 --- a/internal/app/apollo/service.go +++ b/internal/app/apollo/service.go @@ -29,6 +29,10 @@ const ( RecurringEvery = time.Hour // enrichChunk is how many contacts an enrichment run loads and saves at once. enrichChunk = 100 + // runTimeout bounds one import run; staleAfter is when a run still marked + // running is treated as interrupted, and must stay above runTimeout. + runTimeout = 25 * time.Minute + staleAfter = 30 * time.Minute ) // Deps is everything the Apollo integration talks to. @@ -383,7 +387,7 @@ func (s *Service) StartRun(ctx context.Context, orgID, id uuid.UUID) (*models.Ap return nil, errx.NewWithIdentifier(errx.Conflict, "import_running", "This import is already running.") } go func(src models.ApolloImportSource) { - bg, cancel := context.WithTimeout(context.Background(), 25*time.Minute) + bg, cancel := context.WithTimeout(context.Background(), runTimeout) defer cancel() s.runSource(bg, src) }(*src) @@ -393,6 +397,9 @@ func (s *Service) StartRun(ctx context.Context, orgID, id uuid.UUID) (*models.Ap // RunRecurring runs every "keep in sync" source that is due. func (s *Service) RunRecurring(ctx context.Context) error { + if _, err := s.Repo.ReleaseStaleSources(ctx, staleAfter); err != nil { + log.Warn().Err(err).Msg("apollo: could not release interrupted imports") + } due, err := s.Repo.DueRecurringSources(ctx, RecurringEvery) if err != nil { return err @@ -405,7 +412,9 @@ func (s *Service) RunRecurring(ctx context.Context) error { if err != nil || !claimed { continue } - s.runSource(ctx, src) + run, cancel := context.WithTimeout(ctx, runTimeout) + s.runSource(run, src) + cancel() } return nil } @@ -422,7 +431,10 @@ func (s *Service) runSource(ctx context.Context, src models.ApolloImportSource) } log.Warn().Err(err).Str("source", src.ID.String()).Msg("apollo: import run failed") } - _ = s.Repo.FinishSource(ctx, src.ID, res, msg) + // The outcome is recorded even when the run's own context has ended. + if err := s.Repo.FinishSource(context.WithoutCancel(ctx), src.ID, res, msg); err != nil { + log.Warn().Err(err).Str("source", src.ID.String()).Msg("apollo: could not record import result") + } } // importOnce reads the list, imports people it has not brought in before and diff --git a/internal/app/integration/clay.go b/internal/app/integration/clay.go index 831060fba..815401785 100644 --- a/internal/app/integration/clay.go +++ b/internal/app/integration/clay.go @@ -101,9 +101,10 @@ var ErrConnectionNotFound = errors.New("connection not found") // clayDeliver posts one flat row to a Clay table's webhook source. func (s *service) clayDeliver(ctx context.Context, sub models.IntegrationEventSubscription, secretCfg map[string]any, data map[string]any, msg eventMessage) error { + saved := stringFromMap(secretCfg, "webhook_url") target := configString(sub.Config, "url") if target == "" { - target = stringFromMap(secretCfg, "webhook_url") + target = saved } if target == "" { return errors.New("add the Clay table's webhook URL on the Clay page first") @@ -112,8 +113,9 @@ func (s *service) clayDeliver(ctx context.Context, sub models.IntegrationEventSu if err != nil { return err } + // The connection's sealed token goes only to the connection's own table webhook. token := configString(sub.Config, "auth_token") - if token == "" { + if token == "" && target == saved { token = stringFromMap(secretCfg, "auth_token") } return clayPost(ctx, rendered, token, clayRow(sub.EventType, data, msg)) diff --git a/internal/repository/pg_apollo.go b/internal/repository/pg_apollo.go index 8894507c9..5b001c10b 100644 --- a/internal/repository/pg_apollo.go +++ b/internal/repository/pg_apollo.go @@ -40,6 +40,7 @@ type ApolloRepository interface { ClaimSource(ctx context.Context, id uuid.UUID) (bool, error) FinishSource(ctx context.Context, id uuid.UUID, result *models.ApolloRunResult, errMsg string) error DueRecurringSources(ctx context.Context, every time.Duration) ([]models.ApolloImportSource, error) + ReleaseStaleSources(ctx context.Context, after time.Duration) (int64, error) SourceMembers(ctx context.Context, sourceID uuid.UUID, apolloIDs []string) (map[string]bool, error) AddSourceMembers(ctx context.Context, sourceID uuid.UUID, members map[string]uuid.UUID) error @@ -180,6 +181,19 @@ func (r *apolloRepository) ClaimSource(ctx context.Context, id uuid.UUID) (bool, return tag.RowsAffected() > 0, nil } +// ReleaseStaleSources marks runs that outlived any live run (a restart mid-import) +// as interrupted, so they show an error, can be run again and sync on schedule. +func (r *apolloRepository) ReleaseStaleSources(ctx context.Context, after time.Duration) (int64, error) { + tag, err := r.db.Exec(ctx, ` + UPDATE apollo_import_sources + SET status = 'error', last_error = 'The last run was interrupted. It runs again on its next check, or choose Run now.', updated_at = NOW() + WHERE status = 'running' AND updated_at < NOW() - make_interval(secs => $1)`, after.Seconds()) + if err != nil { + return 0, err + } + return tag.RowsAffected(), nil +} + func (r *apolloRepository) FinishSource(ctx context.Context, id uuid.UUID, result *models.ApolloRunResult, errMsg string) error { var raw []byte added := 0 @@ -206,7 +220,8 @@ func (r *apolloRepository) DueRecurringSources(ctx context.Context, every time.D s.last_result, s.last_error, s.total_imported, s.created_at, s.updated_at FROM apollo_import_sources s JOIN integration_connections c ON c.id = s.connection_id - WHERE s.recurring AND s.enabled AND s.status <> 'running' + WHERE s.recurring AND s.enabled + AND (s.status <> 'running' OR s.updated_at < NOW() - INTERVAL '30 minutes') AND c.status IN ('connected', 'degraded') AND (s.last_run_at IS NULL OR s.last_run_at < NOW() - make_interval(secs => $1)) ORDER BY s.last_run_at NULLS FIRST diff --git a/web/src/app/app/integrations/_components/ConnectionDetail.tsx b/web/src/app/app/integrations/_components/ConnectionDetail.tsx index 9d520d370..cd877e836 100644 --- a/web/src/app/app/integrations/_components/ConnectionDetail.tsx +++ b/web/src/app/app/integrations/_components/ConnectionDetail.tsx @@ -227,36 +227,36 @@ export default function ConnectionDetail({ {/* Salesforce has its own page: sync rules, field map, imports, activity log. */} {conn.provider === "salesforce" && (
- } - title="Salesforce settings" - body="Sync rules, field mapping, imports from list views and campaigns, and the activity log." - /> + + } + title="Salesforce settings" + body="Sync rules, field mapping, imports from list views and campaigns, and the activity log." + /> +
)} {conn.provider === "apollo" && (
- } - title="Apollo settings" - body="Imports from your Apollo lists, enrichment runs and the API key." - /> + + } + title="Apollo settings" + body="Imports from your Apollo lists, enrichment runs and the API key." + /> +
)} {conn.provider === "clay" && (
- } - title="Clay settings" - body="The inbound URL and column setup, where rows land, and which events go back to Clay." - /> + + } + title="Clay settings" + body="The inbound URL and column setup, where rows land, and which events go back to Clay." + /> +
)} @@ -368,32 +368,28 @@ export default function ConnectionDetail({ } // A link to a provider's own home page, for settings too rich for this popup. -function HomeLink({ - to, - onNavigate, - icon, - title, - body, -}: { - to: string; - onNavigate: () => void; - icon: React.ReactNode; - title: string; - body: string; -}) { +const homeLinkClass = + "flex items-center gap-3 rounded-md border border-slate-200 hover:border-slate-300 hover:bg-slate-50/60 px-3 py-2.5 transition-colors group"; + +// A link to a provider's own page; typed routes wrap this in their own . +function HomeLink({ to, onNavigate, ...body }: { to: string; onNavigate: () => void } & React.ComponentProps) { return ( - + + + + ); +} + +function HomeLinkBody({ icon, title, body }: { icon: React.ReactNode; title: string; body: string }) { + return ( + <> {icon} {title} {body} - + ); } diff --git a/web/src/app/app/integrations/page.tsx b/web/src/app/app/integrations/page.tsx index 2dca47a09..04d0a1fa8 100644 --- a/web/src/app/app/integrations/page.tsx +++ b/web/src/app/app/integrations/page.tsx @@ -486,16 +486,16 @@ function IntegrationsStore({ route, routeKey }: { route: Route; routeKey: string if (conn.provider === "clay") { // The Clay page shows the new URL with the column setup next to it. if (conn.inbound_webhook_url) stashInboundUrl(conn.id, conn.inbound_webhook_url); - go(`${STORE_BASE}/clay/${conn.id}`); + navigate({ to: "/app/integrations/clay/$id", params: { id: conn.id } }); } else if (conn.provider === "apollo") { - go(`${STORE_BASE}/apollo/${conn.id}`); + navigate({ to: "/app/integrations/apollo/$id", params: { id: conn.id } }); } else if (conn.inbound_webhook_url) { setInboundUrl({ provider: conn.provider, url: conn.inbound_webhook_url }); } else if (CRM_HOMES.has(conn.provider)) { go(`${STORE_BASE}/${conn.provider}`); } else if (conn.provider === "salesforce") { // Setup continues on the Salesforce page: rules, mapping, imports. - go(`${STORE_BASE}/salesforce/${conn.id}`); + navigate({ to: "/app/integrations/salesforce/$id", params: { id: conn.id } }); } else { setManageTarget(conn); } diff --git a/web/src/components/app/automations/AutomationEditors.tsx b/web/src/components/app/automations/AutomationEditors.tsx index 2f50a26cd..fc098ae5d 100644 --- a/web/src/components/app/automations/AutomationEditors.tsx +++ b/web/src/components/app/automations/AutomationEditors.tsx @@ -378,7 +378,7 @@ export function ActionEditor({
setConfig("url", v)} placeholder="Uses the webhook saved on the Clay connection" className="w-full" /> -

Set this to send to a different Clay table than the connection default.

+

Sends to a different Clay table than the connection default. The connection's auth token is not sent to it.

)} {actionSupportsTemplate(data.action ?? "") && (