starred/warmbly
1
0
Fork 0
mirror of https://github.com/warmbly/warmbly.git synced 2026-10-13 00:02:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Matthew Meszaros
2026-10-11 12:59:05 +02:00
parent ebfd34fb84
commit f9a9163891
8 changed files with 78 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/content/docs/guides/clay.mdx
+2 -2
View File
@@ -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.
internal/api/handler/clay.go
+1 -1
View File
@@ -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
}
internal/app/apollo/service.go
+15 -3
View File
@@ -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
internal/app/integration/clay.go
+4 -2
View File
@@ -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))
internal/repository/pg_apollo.go
+16 -1
View File
@@ -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
web/src/app/app/integrations/_components/ConnectionDetail.tsx
+36 -40
View File
@@ -227,36 +227,36 @@ export default function ConnectionDetail({
{/* Salesforce has its own page: sync rules, field map, imports, activity log. */}
{conn.provider === "salesforce" && (
<div className="px-5 py-4">
<HomeLink
to={`/app/integrations/salesforce/${conn.id}`}
onNavigate={onClose}
icon={<Settings2Icon className="w-4 h-4 text-sky-600" />}
title="Salesforce settings"
body="Sync rules, field mapping, imports from list views and campaigns, and the activity log."
/>
<Link to="/app/integrations/salesforce/$id" params={{ id: conn.id }} onClick={onClose} className={homeLinkClass}>
<HomeLinkBody
icon={<Settings2Icon className="w-4 h-4 text-sky-600" />}
title="Salesforce settings"
body="Sync rules, field mapping, imports from list views and campaigns, and the activity log."
/>
</Link>
</div>
)}
{conn.provider === "apollo" && (
<div className="px-5 py-4">
<HomeLink
to={`/app/integrations/apollo/${conn.id}`}
onNavigate={onClose}
icon={<Settings2Icon className="w-4 h-4 text-sky-600" />}
title="Apollo settings"
body="Imports from your Apollo lists, enrichment runs and the API key."
/>
<Link to="/app/integrations/apollo/$id" params={{ id: conn.id }} onClick={onClose} className={homeLinkClass}>
<HomeLinkBody
icon={<Settings2Icon className="w-4 h-4 text-sky-600" />}
title="Apollo settings"
body="Imports from your Apollo lists, enrichment runs and the API key."
/>
</Link>
</div>
)}
{conn.provider === "clay" && (
<div className="px-5 py-4">
<HomeLink
to={`/app/integrations/clay/${conn.id}`}
onNavigate={onClose}
icon={<Settings2Icon className="w-4 h-4 text-sky-600" />}
title="Clay settings"
body="The inbound URL and column setup, where rows land, and which events go back to Clay."
/>
<Link to="/app/integrations/clay/$id" params={{ id: conn.id }} onClick={onClose} className={homeLinkClass}>
<HomeLinkBody
icon={<Settings2Icon className="w-4 h-4 text-sky-600" />}
title="Clay settings"
body="The inbound URL and column setup, where rows land, and which events go back to Clay."
/>
</Link>
</div>
)}
@@ -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 <Link>.
function HomeLink({ to, onNavigate, ...body }: { to: string; onNavigate: () => void } & React.ComponentProps<typeof HomeLinkBody>) {
return (
<Link
to={to}
onClick={onNavigate}
className="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"
>
<Link to={to} onClick={onNavigate} className={homeLinkClass}>
<HomeLinkBody {...body} />
</Link>
);
}
function HomeLinkBody({ icon, title, body }: { icon: React.ReactNode; title: string; body: string }) {
return (
<>
<span className="shrink-0">{icon}</span>
<span className="min-w-0 flex-1">
<span className="block text-[12.5px] font-medium text-slate-900">{title}</span>
<span className="block text-[11.5px] text-slate-500 leading-relaxed">{body}</span>
</span>
<ArrowRightIcon className="w-3.5 h-3.5 text-slate-400 group-hover:text-slate-700 shrink-0" />
</Link>
</>
);
}
web/src/app/app/integrations/page.tsx
+3 -3
View File
@@ -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);
}
web/src/components/app/automations/AutomationEditors.tsx
+1 -1
View File
@@ -378,7 +378,7 @@ export function ActionEditor({
<div>
<Label>Clay table webhook (optional)</Label>
<TextInput value={String(config.url ?? "")} onChange={(v) => setConfig("url", v)} placeholder="Uses the webhook saved on the Clay connection" className="w-full" />
<p className="text-[11px] text-slate-400 mt-1">Set this to send to a different Clay table than the connection default.</p>
<p className="text-[11px] text-slate-400 mt-1">Sends to a different Clay table than the connection default. The connection's auth token is not sent to it.</p>
</div>
)}
{actionSupportsTemplate(data.action ?? "") && (