From 60773b3d8e43a1a1b1c3244f0467bdf9f08415fa Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 30 May 2026 04:17:10 +0000 Subject: [PATCH 01/11] feat: make dashboard realtime --- AGENTS.md | 7 + cmd/backend/main.go | 3 +- internal/app/consumer/event_flags_update.go | 19 +- internal/app/consumer/event_new_email.go | 23 ++ internal/app/consumer/event_update_email.go | 11 +- internal/app/contact/handler.go | 38 ++- internal/app/contact/import.go | 3 + internal/app/contact/service.go | 28 ++- internal/app/email/handler.go | 3 + internal/app/email/onboarding.go | 3 + internal/app/email/service.go | 81 ++++-- internal/infrastructure/pubsub/events.go | 40 +++ realtime/lib/realtime/pubsub/subscriber.ex | 8 + .../components/shared/ConnectionIndicator.tsx | 6 +- web/src/hooks/useRealtimeEvents.ts | 233 ++++++++++++++---- 15 files changed, 409 insertions(+), 97 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 7ddfc6cc..646a02d1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -29,8 +29,15 @@ Other CI-touching rules: Commit hygiene: +- when instructed to make a commit, use the subject format `feat: one line explanation` - commit messages on this repo do not include `Co-Authored-By:` or other AI/agent attribution footers. Keep messages to subject + body explaining the why. If a commit slips through with an attribution footer, rewrite it before opening or updating a PR. +Dashboard realtime: + +- dashboard experiences should be realtime by default. When emails arrive, contacts are added, records change, or any dashboard-visible feature updates, the dashboard should reflect it live without requiring a manual refresh +- aim for a responsive, Discord-like product feel: presence, counts, lists, detail panes, notifications, and workflow state should stay current across every dashboard feature where live updates are meaningful +- when changing dashboard behavior, it is acceptable to safely change the API structure if a better solution exists. Before making an API shape change, ask the user how they want to handle it, especially when the current API may already be published or backwards compatibility might require a new API version + ## System Shape - `cmd/backend`: API and business orchestration diff --git a/cmd/backend/main.go b/cmd/backend/main.go index ffab874c..51775e9e 100644 --- a/cmd/backend/main.go +++ b/cmd/backend/main.go @@ -623,6 +623,7 @@ func main() { cache, &oauth2Cfg.InboxAuthorization, workerAssignmentService, + streamingPublisher, ) // Fan out email-account lifecycle events to customer webhooks. emailService.WireWebhooks(webhookService) @@ -632,7 +633,7 @@ func main() { emailService.WireThrottle(dailyThrottleService) campaignService = campaign.NewService(campaignRepostory, taskRepository, emailRepostory, campaignLogRepository, featureGateService, dailyThrottleService, streamingPublisher) sequenceService = sequence.NewService(sequenceRepostory) - contactService = contact.NewService(contactRepostory, subscriptionRepository, planRepository) + contactService = contact.NewService(contactRepostory, subscriptionRepository, planRepository, streamingPublisher) apiKeyService = apikey.NewService(cache, apiKeyRepository) crmService = crm.NewService(crmRepository) socketService = socket.NewService(cache, tokenService) diff --git a/internal/app/consumer/event_flags_update.go b/internal/app/consumer/event_flags_update.go index a3650965..ffa386c7 100644 --- a/internal/app/consumer/event_flags_update.go +++ b/internal/app/consumer/event_flags_update.go @@ -51,7 +51,7 @@ func (s *JobsService) HandleFlagsAdd(ctx context.Context, e *models.JobEventFlag return nil } - return s.UniboxRepository.UpdateEntry( + if err := s.UniboxRepository.UpdateEntry( ctx, e.UserID, e.EmailID, @@ -59,7 +59,12 @@ func (s *JobsService) HandleFlagsAdd(ctx context.Context, e *models.JobEventFlag &repository.UpdateUniboxEntry{ Flags: email.Flags, }, - ) + ); err != nil { + return err + } + + s.publishEmailUpdated(ctx, e.UserID, email) + return nil } func warmupTokenFromFlags(flags []string) string { @@ -106,7 +111,7 @@ func (s *JobsService) HandleFlagsRemove(ctx context.Context, e *models.JobEventF return nil } - return s.UniboxRepository.UpdateEntry( + if err := s.UniboxRepository.UpdateEntry( ctx, e.UserID, e.EmailID, @@ -114,5 +119,11 @@ func (s *JobsService) HandleFlagsRemove(ctx context.Context, e *models.JobEventF &repository.UpdateUniboxEntry{ Flags: newFlags, }, - ) + ); err != nil { + return err + } + + email.Flags = newFlags + s.publishEmailUpdated(ctx, e.UserID, email) + return nil } diff --git a/internal/app/consumer/event_new_email.go b/internal/app/consumer/event_new_email.go index 80c33f47..a8b1a6a8 100644 --- a/internal/app/consumer/event_new_email.go +++ b/internal/app/consumer/event_new_email.go @@ -9,6 +9,7 @@ import ( "github.com/google/uuid" "github.com/warmbly/warmbly/internal/config" + "github.com/warmbly/warmbly/internal/infrastructure/pubsub" "github.com/warmbly/warmbly/internal/models" ) @@ -36,6 +37,9 @@ func (s *JobsService) HandleNewEmail(ctx context.Context, e *models.JobEventNewE CaptureError(e.UserID, e.Message.EmailID, err) return err } + if s.StreamingPublisher != nil && e.Message != nil { + s.StreamingPublisher.PublishEmailReceived(ctx, emailInboxEvent(e.UserID, e.Message)) + } // Advanced reply-intent automation is best-effort and should not block inbox ingest. if s.AdvancedService != nil { @@ -45,6 +49,25 @@ func (s *JobsService) HandleNewEmail(ctx context.Context, e *models.JobEventNewE return nil } +func (s *JobsService) publishEmailUpdated(ctx context.Context, userID uuid.UUID, message *models.EmailMessageStoreData) { + if s.StreamingPublisher == nil || message == nil { + return + } + s.StreamingPublisher.PublishEmailUpdated(ctx, emailInboxEvent(userID, message)) +} + +func emailInboxEvent(userID uuid.UUID, message *models.EmailMessageStoreData) *pubsub.EmailInboxEvent { + return &pubsub.EmailInboxEvent{ + BaseEvent: pubsub.BaseEvent{UserID: userID.String()}, + EmailAccountID: message.EmailID.String(), + MessageID: message.ID.String(), + ThreadID: message.ThreadID, + Subject: message.Subject, + From: strings.Join(message.FromAddr, ", "), + Preview: message.Snippet, + } +} + // extractHeaderValue extracts a custom header value from the email message // Checks InReplyTo field encoding or direct header access func extractHeaderValue(msg *models.EmailMessageStoreData, headerName string) string { diff --git a/internal/app/consumer/event_update_email.go b/internal/app/consumer/event_update_email.go index 6a740fb1..3110a044 100644 --- a/internal/app/consumer/event_update_email.go +++ b/internal/app/consumer/event_update_email.go @@ -31,5 +31,14 @@ func (s *JobsService) HandleUpdateEmail(ctx context.Context, e *models.JobEventE updateData.ModSeq = &e.ModSeq } - return s.UniboxRepository.UpdateEntry(ctx, e.UserID, e.EmailID, e.ID, &updateData) + if err := s.UniboxRepository.UpdateEntry(ctx, e.UserID, e.EmailID, e.ID, &updateData); err != nil { + return err + } + + email.Flags = e.Flags + email.UID = e.UID + email.Mailbox = e.Mailbox + email.ModSeq = e.ModSeq + s.publishEmailUpdated(ctx, e.UserID, email) + return nil } diff --git a/internal/app/contact/handler.go b/internal/app/contact/handler.go index 6e22bf4a..c69b193d 100644 --- a/internal/app/contact/handler.go +++ b/internal/app/contact/handler.go @@ -31,7 +31,13 @@ func (s *contactService) Add(ctx context.Context, userID string, contacts []mode } } - return s.contactRepository.Add(ctx, userID, contacts) + created, xerr := s.contactRepository.Add(ctx, userID, contacts) + if xerr != nil { + return nil, xerr + } + + s.publishContactsReload(ctx, userID, "contacts:add") + return created, nil } func (s *contactService) Search(ctx context.Context, userID, cursor, category, limit string, filters models.SearchContacts) (*models.ContactsResult, *errx.Error) { @@ -53,19 +59,41 @@ func (s *contactService) Search(ctx context.Context, userID, cursor, category, l } func (s *contactService) BulkUpdate(ctx context.Context, userID string, data *models.BulkEditContactsData) ([]models.Contact, *errx.Error) { - return s.contactRepository.BulkUpdate(ctx, userID, data) + updated, xerr := s.contactRepository.BulkUpdate(ctx, userID, data) + if xerr != nil { + return nil, xerr + } + + s.publishContactsReload(ctx, userID, "contacts:bulk_update") + return updated, nil } func (s *contactService) Update(ctx context.Context, userID, contactID string, data *models.UpdateContact) (*models.Contact, *errx.Error) { - return s.contactRepository.Update(ctx, userID, contactID, data) + updated, xerr := s.contactRepository.Update(ctx, userID, contactID, data) + if xerr != nil { + return nil, xerr + } + + s.publishContactsReload(ctx, userID, "contacts:update:"+contactID) + return updated, nil } func (s *contactService) BulkDelete(ctx context.Context, userID string, contactIDs []string) *errx.Error { - return s.contactRepository.BulkDelete(ctx, userID, contactIDs) + if xerr := s.contactRepository.BulkDelete(ctx, userID, contactIDs); xerr != nil { + return xerr + } + + s.publishContactsReload(ctx, userID, "contacts:bulk_delete") + return nil } func (s *contactService) Delete(ctx context.Context, userID string, contactID string) *errx.Error { - return s.contactRepository.Delete(ctx, userID, contactID) + if xerr := s.contactRepository.Delete(ctx, userID, contactID); xerr != nil { + return xerr + } + + s.publishContactsReload(ctx, userID, "contacts:delete:"+contactID) + return nil } func (s *contactService) GetDetail(ctx context.Context, userID uuid.UUID, orgID *uuid.UUID, contactID uuid.UUID) (*models.ContactDetail, *errx.Error) { diff --git a/internal/app/contact/import.go b/internal/app/contact/import.go index 04a0db1a..a8918da0 100644 --- a/internal/app/contact/import.go +++ b/internal/app/contact/import.go @@ -296,6 +296,9 @@ func (s *contactService) ImportCommit( } res.EndedAt = time.Now().UTC() + if res.Imported > 0 || res.Updated > 0 { + s.publishContactsReload(ctx, userID, "contacts:import") + } return res, nil } diff --git a/internal/app/contact/service.go b/internal/app/contact/service.go index 06c3d1ba..5df3250f 100644 --- a/internal/app/contact/service.go +++ b/internal/app/contact/service.go @@ -7,6 +7,7 @@ import ( "github.com/google/uuid" "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/infrastructure/pubsub" "github.com/warmbly/warmbly/internal/models" "github.com/warmbly/warmbly/internal/repository" ) @@ -47,19 +48,34 @@ type ContactService interface { } type contactService struct { - contactRepository repository.ContactRepository - subRepo repository.SubscriptionRepository - planRepo repository.PlanRepository + contactRepository repository.ContactRepository + subRepo repository.SubscriptionRepository + planRepo repository.PlanRepository + streamingPublisher *pubsub.StreamingPublisher } func NewService( contactRepository repository.ContactRepository, subRepo repository.SubscriptionRepository, planRepo repository.PlanRepository, + streamingPublisher ...*pubsub.StreamingPublisher, ) ContactService { + var publisher *pubsub.StreamingPublisher + if len(streamingPublisher) > 0 { + publisher = streamingPublisher[0] + } + return &contactService{ - contactRepository: contactRepository, - subRepo: subRepo, - planRepo: planRepo, + contactRepository: contactRepository, + subRepo: subRepo, + planRepo: planRepo, + streamingPublisher: publisher, } } + +func (s *contactService) publishContactsReload(ctx context.Context, userID string, operationID string) { + if s.streamingPublisher == nil { + return + } + s.streamingPublisher.PublishContactsReload(ctx, userID, operationID) +} diff --git a/internal/app/email/handler.go b/internal/app/email/handler.go index 422fd136..ca442742 100644 --- a/internal/app/email/handler.go +++ b/internal/app/email/handler.go @@ -4,6 +4,7 @@ import ( "context" "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/infrastructure/pubsub" "github.com/warmbly/warmbly/internal/models" "github.com/warmbly/warmbly/internal/utils/validate" ) @@ -41,6 +42,7 @@ func (s *emailService) Update(ctx context.Context, userID, emailAccountID string } s.syncWarmupPoolMembership(ctx, account) + s.publishAccountEvent(ctx, pubsub.EventAccountSynced, account) return account, nil } @@ -59,6 +61,7 @@ func (s *emailService) Delete(ctx context.Context, userID, emailAccountID string } s.removeFromAllWarmupPools(ctx, account) + s.publishAccountEvent(ctx, pubsub.EventAccountDisconnected, account) if s.webhookService != nil && account != nil && account.OrganizationID != nil { _, _ = s.webhookService.Dispatch(ctx, *account.OrganizationID, models.WebhookEventEmailAccountRemoved, map[string]any{ diff --git a/internal/app/email/onboarding.go b/internal/app/email/onboarding.go index 99071795..d1cd12f5 100644 --- a/internal/app/email/onboarding.go +++ b/internal/app/email/onboarding.go @@ -14,6 +14,7 @@ import ( "github.com/warmbly/warmbly/internal/app/dailythrottle" "github.com/warmbly/warmbly/internal/config" "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/infrastructure/pubsub" "github.com/warmbly/warmbly/internal/models" "github.com/warmbly/warmbly/internal/pkg/crypt" "golang.org/x/oauth2" @@ -161,6 +162,7 @@ func (s *emailService) OAuthFinish(ctx context.Context, userID, code, state stri ExpiresAt: tok.Expiry, }) if xerr == nil && acc != nil { + s.publishAccountEvent(ctx, pubsub.EventAccountConnected, acc) s.dispatchAccountConnected(ctx, sess.OrganizationID, acc) } return acc, xerr @@ -222,6 +224,7 @@ func (s *emailService) OnboardSMTPIMAP(ctx context.Context, userID string, orgID } } + s.publishAccountEvent(ctx, pubsub.EventAccountConnected, acc) s.dispatchAccountConnected(ctx, orgID, acc) return acc, nil } diff --git a/internal/app/email/service.go b/internal/app/email/service.go index e763b2f3..2c8b3787 100644 --- a/internal/app/email/service.go +++ b/internal/app/email/service.go @@ -15,6 +15,7 @@ import ( "github.com/warmbly/warmbly/internal/events" "github.com/warmbly/warmbly/internal/infrastructure/cache" "github.com/warmbly/warmbly/internal/infrastructure/kafka" + "github.com/warmbly/warmbly/internal/infrastructure/pubsub" "github.com/warmbly/warmbly/internal/models" "github.com/warmbly/warmbly/internal/repository" ) @@ -38,16 +39,17 @@ type EmailService interface { } type emailService struct { - emailRepository repository.EmailRepository - cipherService cipher.CipherService - featureGate feature.FeatureGateService - warmupService warmupapp.Service - publisher events.Publisher - producer *kafka.Producer - r *cache.Cache - oauthInbox *config.Oauth2Inbox - workerAssignment worker.WorkerAssignmentService - throttle dailythrottle.Service + emailRepository repository.EmailRepository + cipherService cipher.CipherService + featureGate feature.FeatureGateService + warmupService warmupapp.Service + publisher events.Publisher + streamingPublisher *pubsub.StreamingPublisher + producer *kafka.Producer + r *cache.Cache + oauthInbox *config.Oauth2Inbox + workerAssignment worker.WorkerAssignmentService + throttle dailythrottle.Service // webhookService is optional. When non-nil, account lifecycle events // (email_account.connected, email_account.removed) are dispatched to // subscribed customer webhooks. @@ -74,13 +76,20 @@ func NewService( featureGate feature.FeatureGateService, warmupService warmupapp.Service, publisher events.Publisher, + streamingPublisher ...*pubsub.StreamingPublisher, ) EmailService { + var realtime *pubsub.StreamingPublisher + if len(streamingPublisher) > 0 { + realtime = streamingPublisher[0] + } + return &emailService{ - emailRepository: emailRepository, - cipherService: cipherService, - featureGate: featureGate, - warmupService: warmupService, - publisher: publisher, + emailRepository: emailRepository, + cipherService: cipherService, + featureGate: featureGate, + warmupService: warmupService, + publisher: publisher, + streamingPublisher: realtime, } } @@ -94,16 +103,40 @@ func NewServiceWithKafka( r *cache.Cache, oauthInbox *config.Oauth2Inbox, workerAssignment worker.WorkerAssignmentService, + streamingPublisher ...*pubsub.StreamingPublisher, ) EmailService { + var realtime *pubsub.StreamingPublisher + if len(streamingPublisher) > 0 { + realtime = streamingPublisher[0] + } + return &emailService{ - emailRepository: emailRepository, - cipherService: cipherService, - featureGate: featureGate, - warmupService: warmupService, - publisher: publisher, - producer: producer, - r: r, - oauthInbox: oauthInbox, - workerAssignment: workerAssignment, + emailRepository: emailRepository, + cipherService: cipherService, + featureGate: featureGate, + warmupService: warmupService, + publisher: publisher, + streamingPublisher: realtime, + producer: producer, + r: r, + oauthInbox: oauthInbox, + workerAssignment: workerAssignment, } } + +func (s *emailService) publishAccountEvent(ctx context.Context, eventType pubsub.EventType, account *models.Email) { + if s.streamingPublisher == nil || account == nil { + return + } + + s.streamingPublisher.PublishAccountEvent(ctx, &pubsub.AccountEvent{ + BaseEvent: pubsub.BaseEvent{ + EventType: eventType, + UserID: account.UserID, + }, + EmailAccountID: account.ID.String(), + Email: account.Email, + Provider: account.Provider, + Status: account.Status, + }) +} diff --git a/internal/infrastructure/pubsub/events.go b/internal/infrastructure/pubsub/events.go index 5c7d5244..5c3cf073 100644 --- a/internal/infrastructure/pubsub/events.go +++ b/internal/infrastructure/pubsub/events.go @@ -184,6 +184,46 @@ func (p *StreamingPublisher) PublishEmailReceived(ctx context.Context, event *Em } } +// PublishEmailUpdated notifies user that an inbox row changed. +func (p *StreamingPublisher) PublishEmailUpdated(ctx context.Context, event *EmailInboxEvent) { + if p.client == nil { + return + } + + event.EventType = EventEmailUpdated + event.Timestamp = time.Now() + + attrs := map[string]string{ + "user_id": event.UserID, + "email_id": event.EmailAccountID, + "event_type": string(EventEmailUpdated), + } + + if err := p.client.Publish(ctx, TopicEmailInbox, event, attrs); err != nil { + // Log error but don't fail + } +} + +// PublishEmailDeleted notifies user that an inbox row was removed. +func (p *StreamingPublisher) PublishEmailDeleted(ctx context.Context, event *EmailInboxEvent) { + if p.client == nil { + return + } + + event.EventType = EventEmailDeleted + event.Timestamp = time.Now() + + attrs := map[string]string{ + "user_id": event.UserID, + "email_id": event.EmailAccountID, + "event_type": string(EventEmailDeleted), + } + + if err := p.client.Publish(ctx, TopicEmailInbox, event, attrs); err != nil { + // Log error but don't fail + } +} + // PublishContactsReload signals frontend to reload contacts func (p *StreamingPublisher) PublishContactsReload(ctx context.Context, userID, operationID string) { if p.client == nil { diff --git a/realtime/lib/realtime/pubsub/subscriber.ex b/realtime/lib/realtime/pubsub/subscriber.ex index 633fc318..0b0ffcaf 100644 --- a/realtime/lib/realtime/pubsub/subscriber.ex +++ b/realtime/lib/realtime/pubsub/subscriber.ex @@ -86,6 +86,14 @@ defmodule Realtime.CloudPubSub.Subscriber do Logger.debug("Broadcast #{event_type} to #{topic}") end + org_id = event["org_id"] || event["organization_id"] + + if org_id do + topic = "org:#{org_id}" + Phoenix.PubSub.broadcast(Realtime.PubSub, topic, {:pubsub_event, event}) + Logger.debug("Broadcast #{event_type} to #{topic}") + end + # Broadcast to entity-specific channels broadcast_to_entity_channels(event) end diff --git a/web/src/components/shared/ConnectionIndicator.tsx b/web/src/components/shared/ConnectionIndicator.tsx index ad337c98..c5ef5ad8 100644 --- a/web/src/components/shared/ConnectionIndicator.tsx +++ b/web/src/components/shared/ConnectionIndicator.tsx @@ -4,15 +4,12 @@ import { cn } from '@/lib/utils' export function ConnectionIndicator() { const { status, quality } = useConnectionStatus() - if (status === 'connected' && quality === 'good') { - return null - } - return (
+ {status === 'connected' && quality === 'good' && 'Live'} {status === 'connecting' && 'Reconnecting...'} {status === 'disconnected' && 'Disconnected'} {status === 'connected' && quality === 'degraded' && 'Slow connection'} diff --git a/web/src/hooks/useRealtimeEvents.ts b/web/src/hooks/useRealtimeEvents.ts index 4406fa19..a1cfa599 100644 --- a/web/src/hooks/useRealtimeEvents.ts +++ b/web/src/hooks/useRealtimeEvents.ts @@ -1,4 +1,5 @@ -import { useEffect } from 'react' +import { useCallback, useEffect } from 'react' +import { useQueryClient, type QueryKey } from '@tanstack/react-query' import { useSocket } from './context/socket' import { useAppStore } from '@/stores' import { useUserProfile } from './context/user' @@ -6,76 +7,204 @@ import { useUserProfile } from './context/user' export function useRealtimeEvents() { const { isConnected, subscribeToChannel } = useSocket() const { user } = useUserProfile() + const queryClient = useQueryClient() const currentOrg = useAppStore((s) => s.currentOrganization) const updateCampaign = useAppStore((s) => s.updateCampaign) - const addUniboxEmail = useAppStore((s) => s.addUniboxEmail) const incrementUnseenCount = useAppStore((s) => s.incrementUnseenCount) const updateDeal = useAppStore((s) => s.updateDeal) const setSubscription = useAppStore((s) => s.setSubscription) - // User channel events - useEffect(() => { - if (!isConnected || !user?.email) return + const invalidate = useCallback( + (queryKeys: QueryKey[]) => { + for (const queryKey of queryKeys) { + void queryClient.invalidateQueries({ queryKey }) + } + }, + [queryClient], + ) - const topic = `user:${user.email}` - const unsubs: (() => void)[] = [] + const handleRealtimeEvent = useCallback( + (payload: Record) => { + const rawEvent = String( + payload.event_type ?? payload.type ?? payload._event ?? '', + ) + const event = rawEvent.replace(/[.:\s-]+/g, '_').toUpperCase() + if (!event) return - // New email received - unsubs.push( - subscribeToChannel(topic, 'new_email', (payload) => { - addUniboxEmail(payload as any) + const getString = (key: string) => { + const value = payload[key] + return typeof value === 'string' && value.length > 0 ? value : null + } + const includes = (...needles: string[]) => + needles.some((needle) => event.includes(needle)) + + const campaignId = getString('campaign_id') + const contactId = getString('contact_id') + const dealId = getString('deal_id') + const threadId = getString('thread_id') + const emailId = getString('email_id') ?? getString('message_id') + + if (includes('EMAIL_RECEIVED', 'NEW_EMAIL', 'INBOX_NEW')) { incrementUnseenCount() - }) - ) + invalidate([ + ['unibox'], + ['analytics'], + ['emails', 'list'], + ]) + if (threadId) invalidate([['unibox', 'thread', threadId]]) + if (emailId) invalidate([['unibox', 'email', emailId]]) + return + } - // Campaign status changed - unsubs.push( - subscribeToChannel(topic, 'campaign_status_changed', (payload) => { - const { campaign_id, ...updates } = payload as any - if (campaign_id) { - updateCampaign(campaign_id, updates) + if (includes('EMAIL_UPDATED', 'EMAIL_DELETED', 'INBOX_UPDATE')) { + invalidate([['unibox'], ['analytics']]) + if (threadId) invalidate([['unibox', 'thread', threadId]]) + if (emailId) invalidate([['unibox', 'email', emailId]]) + return + } + + if (includes('CONTACT')) { + invalidate([ + ['contacts'], + ['campaigns', 'list'], + ['analytics'], + ['organizations', 'limits'], + ]) + if (contactId) invalidate([['contacts', contactId]]) + return + } + + if ( + includes( + 'CAMPAIGN', + 'EMAIL_SENT', + 'EMAIL_OPENED', + 'EMAIL_CLICKED', + 'EMAIL_REPLIED', + 'EMAIL_BOUNCED', + 'TASK_PROGRESS', + ) + ) { + if (campaignId) { + const status = getString('status') + updateCampaign(campaignId, status ? { status } : {}) + invalidate([ + ['campaigns', campaignId], + ['campaigns', campaignId, 'logs'], + ['analytics', 'campaigns', campaignId], + ['analytics', 'campaigns', campaignId, 'daily'], + ['analytics', 'campaigns', campaignId, 'hourly'], + ]) } - }) - ) + invalidate([ + ['campaigns', 'list'], + ['analytics'], + ['contacts'], + ]) + if (contactId) invalidate([['contacts', contactId]]) + return + } - // Subscription changed - unsubs.push( - subscribeToChannel(topic, 'subscription_changed', (payload) => { + if (includes('ACCOUNT', 'EMAIL_STATUS', 'EMAIL_ERROR', 'WARMUP')) { + invalidate([ + ['emails', 'list'], + ['analytics', 'accounts'], + ['analytics', 'warmup'], + ['analytics', 'dashboard'], + ]) + return + } + + if (includes('DEAL')) { + if (dealId) updateDeal(dealId, payload as any) + invalidate([['crm', 'deals'], ['crm', 'pipelines'], ['contacts']]) + return + } + + if (includes('PIPELINE', 'STAGE')) { + invalidate([['crm', 'pipelines'], ['crm', 'deals']]) + return + } + + if (includes('CRM_TASK', 'TASK')) { + invalidate([['crm', 'tasks'], ['crm', 'deals']]) + return + } + + if (includes('SUBSCRIPTION', 'PLAN', 'BILLING', 'LIMIT')) { setSubscription(payload as any) - }) - ) + invalidate([ + ['subscription'], + ['organizations', 'current'], + ['organizations', 'limits'], + ['auth', 'me'], + ]) + return + } - return () => unsubs.forEach((fn) => fn()) - }, [isConnected, user?.email, subscribeToChannel, addUniboxEmail, incrementUnseenCount, updateCampaign, setSubscription]) + if (includes('MEMBER', 'INVITATION', 'ORGANIZATION', 'SETTINGS')) { + invalidate([ + ['organizations'], + ['organizations', 'current'], + ['organizations', 'invitations'], + ['auth', 'me'], + ]) + return + } + + if (includes('API_KEY')) { + invalidate([['api-keys']]) + return + } + + if (includes('TEMPLATE')) { + invalidate([['templates']]) + return + } + + if (includes('AUDIT')) { + invalidate([['audit']]) + return + } + + if (includes('DANGER', 'DELETION')) { + invalidate([ + ['dangerzone'], + ['auth', 'me'], + ['organizations', 'current'], + ]) + return + } + + invalidate([ + ['analytics', 'dashboard'], + ['auth', 'me'], + ]) + }, + [ + incrementUnseenCount, + invalidate, + setSubscription, + updateCampaign, + updateDeal, + ], + ) + + // User channel events. Topic uses the user UUID; the realtime server + // rejects email-address topics and only authorizes `user:{sub}`. + useEffect(() => { + if (!isConnected || !user?.id) return + + const topic = `user:${user.id}` + return subscribeToChannel(topic, '*', handleRealtimeEvent) + }, [isConnected, user?.id, subscribeToChannel, handleRealtimeEvent]) // Org channel events useEffect(() => { if (!isConnected || !currentOrg?.id) return const topic = `org:${currentOrg.id}` - const unsubs: (() => void)[] = [] - - // Deal updated - unsubs.push( - subscribeToChannel(topic, 'deal_updated', (payload) => { - const { deal_id, ...updates } = payload as any - if (deal_id) { - updateDeal(deal_id, updates) - } - }) - ) - - // Campaign events - unsubs.push( - subscribeToChannel(topic, 'email_sent', (payload) => { - const { campaign_id } = payload as any - if (campaign_id) { - updateCampaign(campaign_id, {}) - } - }) - ) - - return () => unsubs.forEach((fn) => fn()) - }, [isConnected, currentOrg?.id, subscribeToChannel, updateDeal, updateCampaign]) + return subscribeToChannel(topic, '*', handleRealtimeEvent) + }, [isConnected, currentOrg?.id, subscribeToChannel, handleRealtimeEvent]) } From 67bbd7277794eae91ad81b3e77ce8b4205102d76 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 30 May 2026 04:24:41 +0000 Subject: [PATCH 02/11] feat: harden api permission gates --- internal/api/handler/grouph/init.go | 5 +++- internal/api/routes.go | 38 +++++++++++++++++++---------- internal/models/api_permission.go | 12 +++++++-- 3 files changed, 39 insertions(+), 16 deletions(-) diff --git a/internal/api/handler/grouph/init.go b/internal/api/handler/grouph/init.go index c9bb40b2..994673d7 100644 --- a/internal/api/handler/grouph/init.go +++ b/internal/api/handler/grouph/init.go @@ -10,13 +10,16 @@ type Handler struct { service group.GroupService } -func New(r *gin.RouterGroup, service group.GroupService, name string) { +func New(r *gin.RouterGroup, service group.GroupService, name string, middleware ...gin.HandlerFunc) { h := &Handler{ name: name, service: service, } g := r.Group("/" + name) + if len(middleware) > 0 { + g.Use(middleware...) + } { g.POST("", h.Create) g.PATCH("/:gid", h.Update) diff --git a/internal/api/routes.go b/internal/api/routes.go index 4e8bf49f..24f8f308 100644 --- a/internal/api/routes.go +++ b/internal/api/routes.go @@ -61,10 +61,23 @@ func Run( } corsConfig := cors.Config{ - AllowMethods: []string{"POST", "GET", "PATCH", "OPTIONS", "DELETE"}, - AllowHeaders: []string{"Origin", "Content-Type", "Authorization"}, - ExposeHeaders: []string{"Content-Length"}, - MaxAge: 12 * time.Hour, + AllowMethods: []string{"POST", "GET", "PUT", "PATCH", "OPTIONS", "DELETE"}, + AllowHeaders: []string{ + "Origin", + "Content-Type", + "Authorization", + "Idempotency-Key", + "X-Request-Id", + }, + ExposeHeaders: []string{ + "Content-Length", + "X-Request-Id", + "X-RateLimit-Limit", + "X-RateLimit-Remaining", + "X-RateLimit-Policy", + "Retry-After", + }, + MaxAge: 12 * time.Hour, } switch { case len(allowedOrigins) == 0 && ginMode != gin.ReleaseMode: @@ -217,12 +230,11 @@ func Run( contacts.GET("/:id/deals", m.RequireAccess(models.PermViewContacts, models.APIPermReadCRM), h.GetDealsByContact) } - // Group endpoints (folders / tags / categories) don't yet have - // dedicated permission bits — gate them on the broadest read scope - // for now so an API key needs at least one collection permission. - grouph.New(protected, h.FolderService, "folders") - grouph.New(protected, h.TagService, "tags") - grouph.New(protected, h.CategoryService, "categories") + // Group endpoints map to the resources they organize: campaign + // folders, email-account tags, and contact categories. + grouph.New(protected, h.FolderService, "folders", m.RequireAccess(models.PermManageCampaigns, models.APIPermWriteCampaigns)) + grouph.New(protected, h.TagService, "tags", m.RequireAccess(models.PermManageEmails, models.APIPermWriteEmails)) + grouph.New(protected, h.CategoryService, "categories", m.RequireAccess(models.PermManageContacts, models.APIPermWriteContacts)) unibox := protected.Group("/unibox") unibox.Use(m.RateLimitMiddleware(models.RateLimitRead)) @@ -312,7 +324,7 @@ func Run( // Customer-facing webhooks (org-scoped). webhooks := protected.Group("/webhooks") - webhooks.Use(m.RequireOrganization(), m.RateLimitMiddleware(models.RateLimitWrite)) + webhooks.Use(m.RequireOrganization(), m.RequireAccess(models.PermManageSettings, models.APIPermWebhooks), m.RateLimitMiddleware(models.RateLimitWrite)) { webhooks.GET("", h.ListWebhookEndpoints) webhooks.POST("", h.CreateWebhookEndpoint) @@ -326,7 +338,7 @@ func Run( // "available integrations" list; connections are this org's live // state for each provider. integrations := protected.Group("/integrations") - integrations.Use(m.RequireOrganization(), m.RateLimitMiddleware(models.RateLimitWrite)) + integrations.Use(m.RequireOrganization(), m.RequireAccess(models.PermManageSettings, models.APIPermIntegrations), m.RateLimitMiddleware(models.RateLimitWrite)) { integrations.GET("/catalog", h.ListIntegrationCatalog) integrations.GET("/connections", h.ListIntegrationConnections) @@ -339,7 +351,7 @@ func Run( // preferences for premium-pool partner selection — e.g. send // to Gmail recipients only from Google-classified senders. warmupRouting := protected.Group("/warmup/routing") - warmupRouting.Use(m.RequireOrganization(), m.RateLimitMiddleware(models.RateLimitWrite)) + warmupRouting.Use(m.RequireOrganization(), m.RequireAccess(models.PermManageSettings, models.APIPermWarmupRouting), m.RateLimitMiddleware(models.RateLimitWrite)) { warmupRouting.GET("", h.ListWarmupRoutingRules) warmupRouting.POST("", h.CreateWarmupRoutingRule) diff --git a/internal/models/api_permission.go b/internal/models/api_permission.go index 45f7baab..2e06320f 100644 --- a/internal/models/api_permission.go +++ b/internal/models/api_permission.go @@ -43,6 +43,10 @@ const ( // Audit trail APIPermReadAuditLogs + + // Organization operations + APIPermIntegrations // Connect and manage third-party integrations + APIPermWarmupRouting // Manage warmup routing rules ) // AllAPIPermissionsMask is the OR of every defined permission bit. @@ -58,7 +62,8 @@ const AllAPIPermissionsMask uint64 = APIPermReadEmails | APIPermReadCampaigns | APIPermSendCampaigns | APIPermReadTemplates | APIPermWriteTemplates | APIPermReadCRM | APIPermWriteCRM | - APIPermReadAuditLogs + APIPermReadAuditLogs | + APIPermIntegrations | APIPermWarmupRouting // Preset permission sets surfaced via GET /api-keys/permissions so a // caller can grant a sane default without picking bits by hand. @@ -73,7 +78,8 @@ var ( APIPermBulkContacts | APIPermBulkCampaigns | APIPermSendCampaigns | APIPermWriteTemplates | APIPermWriteCRM | - APIPermRealtimeSubscribe | APIPermWebhooks | APIPermAPIKeys + APIPermRealtimeSubscribe | APIPermWebhooks | APIPermAPIKeys | + APIPermIntegrations | APIPermWarmupRouting ) type APIPermission struct { @@ -104,6 +110,8 @@ var AllAPIPermissions = []APIPermission{ {"REALTIME_SUBSCRIBE", APIPermRealtimeSubscribe, "Subscribe to realtime events", "special"}, {"WEBHOOKS", APIPermWebhooks, "Manage webhook endpoints", "special"}, {"API_KEYS", APIPermAPIKeys, "Create and manage API keys", "special"}, + {"INTEGRATIONS", APIPermIntegrations, "Connect and manage third-party integrations", "special"}, + {"WARMUP_ROUTING", APIPermWarmupRouting, "Manage warmup routing rules", "special"}, } // HasAPIPermission reports whether the bitmask grants every bit in `required`. From 3076cff5d62e8938e8261fce766974f3088b1d10 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 30 May 2026 04:26:12 +0000 Subject: [PATCH 03/11] feat: add api request ids --- internal/api/middleware/apikey.go | 2 + internal/api/middleware/request_id.go | 47 ++++++++++++++++++ internal/api/middleware/request_id_test.go | 55 ++++++++++++++++++++++ internal/api/routes.go | 1 + internal/errx/codes.go | 13 +++++ internal/errx/errx.go | 18 ++++--- internal/errx/errx_test.go | 37 +++++++++++++++ 7 files changed, 167 insertions(+), 6 deletions(-) create mode 100644 internal/api/middleware/request_id.go create mode 100644 internal/api/middleware/request_id_test.go create mode 100644 internal/errx/errx_test.go diff --git a/internal/api/middleware/apikey.go b/internal/api/middleware/apikey.go index b7dddd6e..efbbb189 100644 --- a/internal/api/middleware/apikey.go +++ b/internal/api/middleware/apikey.go @@ -96,6 +96,8 @@ func (h *Handler) validateAPIKey(c *gin.Context, rawKey string) { c.AbortWithStatusJSON(http.StatusTooManyRequests, gin.H{ "error": "rate_limit_exceeded", "message": fmt.Sprintf("API key exceeded %d requests per minute", limit), + "code": "rate_limit_exceeded", + "request_id": c.GetString(RequestIDContextKey), "retry_after": retryAfter, }) return diff --git a/internal/api/middleware/request_id.go b/internal/api/middleware/request_id.go new file mode 100644 index 00000000..3682fda7 --- /dev/null +++ b/internal/api/middleware/request_id.go @@ -0,0 +1,47 @@ +package middleware + +import ( + "strings" + + "github.com/gin-gonic/gin" + "github.com/google/uuid" +) + +const ( + RequestIDContextKey = "request_id" + RequestIDHeader = "X-Request-Id" +) + +// RequestIDMiddleware attaches a stable request ID to every request and +// response. Clients may provide one for cross-system tracing; invalid or +// oversized values are replaced so logs and responses stay safe. +func RequestIDMiddleware() gin.HandlerFunc { + return func(c *gin.Context) { + requestID := sanitizeRequestID(c.GetHeader(RequestIDHeader)) + if requestID == "" { + requestID = uuid.NewString() + } + + c.Set(RequestIDContextKey, requestID) + c.Header(RequestIDHeader, requestID) + c.Next() + } +} + +func sanitizeRequestID(value string) string { + value = strings.TrimSpace(value) + if value == "" || len(value) > 128 { + return "" + } + for _, r := range value { + switch { + case r >= 'a' && r <= 'z': + case r >= 'A' && r <= 'Z': + case r >= '0' && r <= '9': + case r == '-' || r == '_' || r == '.' || r == ':': + default: + return "" + } + } + return value +} diff --git a/internal/api/middleware/request_id_test.go b/internal/api/middleware/request_id_test.go new file mode 100644 index 00000000..30c73b06 --- /dev/null +++ b/internal/api/middleware/request_id_test.go @@ -0,0 +1,55 @@ +package middleware + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/gin-gonic/gin" +) + +func TestRequestIDMiddlewareUsesClientRequestID(t *testing.T) { + gin.SetMode(gin.TestMode) + r := gin.New() + r.Use(RequestIDMiddleware()) + r.GET("/x", func(c *gin.Context) { + c.String(http.StatusOK, c.GetString(RequestIDContextKey)) + }) + + req := httptest.NewRequest(http.MethodGet, "/x", nil) + req.Header.Set(RequestIDHeader, "client-trace_123") + rec := httptest.NewRecorder() + r.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d", rec.Code, http.StatusOK) + } + if got := rec.Header().Get(RequestIDHeader); got != "client-trace_123" { + t.Fatalf("response request id = %q", got) + } + if got := rec.Body.String(); got != "client-trace_123" { + t.Fatalf("context request id = %q", got) + } +} + +func TestRequestIDMiddlewareReplacesUnsafeRequestID(t *testing.T) { + gin.SetMode(gin.TestMode) + r := gin.New() + r.Use(RequestIDMiddleware()) + r.GET("/x", func(c *gin.Context) { + c.String(http.StatusOK, c.GetString(RequestIDContextKey)) + }) + + req := httptest.NewRequest(http.MethodGet, "/x", nil) + req.Header.Set(RequestIDHeader, "bad/request/id") + rec := httptest.NewRecorder() + r.ServeHTTP(rec, req) + + got := rec.Header().Get(RequestIDHeader) + if got == "" || got == "bad/request/id" { + t.Fatalf("response request id = %q", got) + } + if got != rec.Body.String() { + t.Fatalf("header request id %q does not match context %q", got, rec.Body.String()) + } +} diff --git a/internal/api/routes.go b/internal/api/routes.go index 24f8f308..66912edc 100644 --- a/internal/api/routes.go +++ b/internal/api/routes.go @@ -22,6 +22,7 @@ func Run( gin.SetMode(ginMode) r := gin.Default() + r.Use(middleware.RequestIDMiddleware()) r.GET("/health", func(c *gin.Context) { c.JSON(200, gin.H{"status": "ok"}) diff --git a/internal/errx/codes.go b/internal/errx/codes.go index 3c3a7190..7e32a1cc 100644 --- a/internal/errx/codes.go +++ b/internal/errx/codes.go @@ -42,3 +42,16 @@ var codeToString = map[Code]string{ NotImplemented: "Not Implemented", ServiceUnavailable: "Service Unavailable", } + +var codeToIdentifier = map[Code]string{ + BadRequest: "bad_request", + Unauthorized: "unauthorized", + Forbidden: "forbidden", + NotFound: "not_found", + Conflict: "conflict", + Unprocessable: "unprocessable", + TooManyRequests: "rate_limit_exceeded", + Internal: "internal_error", + NotImplemented: "not_implemented", + ServiceUnavailable: "service_unavailable", +} diff --git a/internal/errx/errx.go b/internal/errx/errx.go index d0d5f1ef..75bef247 100644 --- a/internal/errx/errx.go +++ b/internal/errx/errx.go @@ -35,8 +35,10 @@ var ( // --- Gin handler helper --- type response struct { - Error string `json:"error"` - Message string `json:"message"` + Error string `json:"error"` + Message string `json:"message"` + Code string `json:"code"` + RequestID string `json:"request_id,omitempty"` } func InternalError() *Error { @@ -50,8 +52,10 @@ func Handle(c *gin.Context, err error) { httpCode := codeToHTTP[bizErr.Code] httpError := codeToString[bizErr.Code] c.JSON(httpCode, response{ - Error: httpError, - Message: bizErr.Message, + Error: httpError, + Message: bizErr.Message, + Code: codeToIdentifier[bizErr.Code], + RequestID: c.GetString("request_id"), }) return } @@ -65,7 +69,9 @@ func JSON(c *gin.Context, err *Error) { httpCode := codeToHTTP[err.Code] httpError := codeToString[err.Code] c.JSON(httpCode, response{ - Error: httpError, - Message: err.Message, + Error: httpError, + Message: err.Message, + Code: codeToIdentifier[err.Code], + RequestID: c.GetString("request_id"), }) } diff --git a/internal/errx/errx_test.go b/internal/errx/errx_test.go new file mode 100644 index 00000000..605308d5 --- /dev/null +++ b/internal/errx/errx_test.go @@ -0,0 +1,37 @@ +package errx + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gin-gonic/gin" +) + +func TestJSONIncludesStableCodeAndRequestID(t *testing.T) { + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Set("request_id", "req_test_123") + + JSON(c, New(BadRequest, "invalid cursor")) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d", rec.Code, http.StatusBadRequest) + } + + var body response + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatalf("decode response: %v", err) + } + if body.Code != "bad_request" { + t.Fatalf("code = %q", body.Code) + } + if body.RequestID != "req_test_123" { + t.Fatalf("request_id = %q", body.RequestID) + } + if body.Error != "Bad Request" || body.Message != "invalid cursor" { + t.Fatalf("unexpected body: %+v", body) + } +} From 878e8d921e2ea9aaa701aa449f1a6a607064cd05 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 30 May 2026 04:28:25 +0000 Subject: [PATCH 04/11] feat: enforce api key email scopes --- internal/api/handler/email.go | 2 +- internal/api/middleware/apikey.go | 63 +++++++++++++++++++++++--- internal/api/middleware/apikey_test.go | 46 +++++++++++++++++++ internal/api/routes.go | 10 ++-- internal/app/analytics/service.go | 2 +- internal/app/email/handler.go | 5 +- internal/app/email/service.go | 2 +- internal/repository/pg_email.go | 12 ++++- 8 files changed, 124 insertions(+), 18 deletions(-) diff --git a/internal/api/handler/email.go b/internal/api/handler/email.go index 1c88fe3b..06866916 100644 --- a/internal/api/handler/email.go +++ b/internal/api/handler/email.go @@ -18,7 +18,7 @@ func (h *Handler) EmailsSearch(c *gin.Context) { tag := c.Query("tag") limit := c.Query("limit") - resp, err := h.EmailService.Search(c.Request.Context(), userID, query, cursor, tag, limit) + resp, err := h.EmailService.Search(c.Request.Context(), userID, query, cursor, tag, limit, middleware.GetAPIKeyAllowedEmailAccounts(c)) if err != nil { errx.Handle(c, err) return diff --git a/internal/api/middleware/apikey.go b/internal/api/middleware/apikey.go index efbbb189..2129202a 100644 --- a/internal/api/middleware/apikey.go +++ b/internal/api/middleware/apikey.go @@ -13,12 +13,13 @@ import ( ) const ( - APIKeyIDKey = "api_key_id" - APIKeyPermissionsKey = "api_key_permissions" - APIKeyUserIDKey = "api_key_user_id" - AuthTypeKey = "auth_type" - AuthTypeJWT = "jwt" - AuthTypeAPIKey = "api_key" + APIKeyIDKey = "api_key_id" + APIKeyPermissionsKey = "api_key_permissions" + APIKeyAllowedEmailAccountsKey = "api_key_allowed_email_accounts" + APIKeyUserIDKey = "api_key_user_id" + AuthTypeKey = "auth_type" + AuthTypeJWT = "jwt" + AuthTypeAPIKey = "api_key" ) // APIKeyMiddleware accepts only API key auth ("Bearer wmbly_..."). Reserved @@ -106,6 +107,7 @@ func (h *Handler) validateAPIKey(c *gin.Context, rawKey string) { c.Set(AuthTypeKey, AuthTypeAPIKey) c.Set(APIKeyIDKey, key.ID.String()) c.Set(APIKeyPermissionsKey, key.Permissions) + c.Set(APIKeyAllowedEmailAccountsKey, key.AllowedEmailAccounts) c.Set(UserIDKey, key.UserID.String()) c.Set(OrganizationIDKey, key.OrganizationID) @@ -215,6 +217,41 @@ func (h *Handler) RequireAccess(orgPerm models.OrganizationPermission, apiPerm u } } +// RequireAPIKeyEmailAccountParam enforces an API key's optional +// allowed_email_accounts allowlist against a route parameter. JWT callers and +// unrestricted API keys pass through. +func RequireAPIKeyEmailAccountParam(param string) gin.HandlerFunc { + return func(c *gin.Context) { + if c.GetString(AuthTypeKey) != AuthTypeAPIKey { + c.Next() + return + } + + allowed := GetAPIKeyAllowedEmailAccounts(c) + if len(allowed) == 0 { + c.Next() + return + } + + accountID, err := uuid.Parse(c.Param(param)) + if err != nil { + errx.Handle(c, errx.ErrUuid) + c.Abort() + return + } + + for _, id := range allowed { + if id == accountID { + c.Next() + return + } + } + + errx.Handle(c, errx.New(errx.Forbidden, "email account is not allowed for this API key")) + c.Abort() + } +} + // GetAuthType returns "jwt" or "api_key" (empty if unauthenticated). func GetAuthType(c *gin.Context) string { return c.GetString(AuthTypeKey) @@ -247,3 +284,17 @@ func GetAPIKeyPermissions(c *gin.Context) uint64 { } return permissions } + +// GetAPIKeyAllowedEmailAccounts returns the optional email-account allowlist +// attached to the authenticating API key. Empty means unrestricted. +func GetAPIKeyAllowedEmailAccounts(c *gin.Context) []uuid.UUID { + value, exists := c.Get(APIKeyAllowedEmailAccountsKey) + if !exists { + return nil + } + ids, ok := value.([]uuid.UUID) + if !ok { + return nil + } + return ids +} diff --git a/internal/api/middleware/apikey_test.go b/internal/api/middleware/apikey_test.go index a43500eb..87c298f5 100644 --- a/internal/api/middleware/apikey_test.go +++ b/internal/api/middleware/apikey_test.go @@ -6,6 +6,7 @@ import ( "testing" "github.com/gin-gonic/gin" + "github.com/google/uuid" "github.com/warmbly/warmbly/internal/models" ) @@ -13,6 +14,51 @@ func init() { gin.SetMode(gin.TestMode) } +func TestRequireAPIKeyEmailAccountParam(t *testing.T) { + gin.SetMode(gin.TestMode) + allowedID := uuid.New() + deniedID := uuid.New() + + tests := []struct { + name string + authType string + allowlist []uuid.UUID + pathID uuid.UUID + wantStatus int + }{ + {"jwt bypasses allowlist", AuthTypeJWT, []uuid.UUID{allowedID}, deniedID, http.StatusOK}, + {"unrestricted key bypasses allowlist", AuthTypeAPIKey, nil, deniedID, http.StatusOK}, + {"allowed key passes", AuthTypeAPIKey, []uuid.UUID{allowedID}, allowedID, http.StatusOK}, + {"denied key fails", AuthTypeAPIKey, []uuid.UUID{allowedID}, deniedID, http.StatusForbidden}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + r := gin.New() + r.Use(RequestIDMiddleware()) + r.GET("/emails/:id", + func(c *gin.Context) { + c.Set(AuthTypeKey, tt.authType) + c.Set(APIKeyAllowedEmailAccountsKey, tt.allowlist) + c.Next() + }, + RequireAPIKeyEmailAccountParam("id"), + func(c *gin.Context) { + c.Status(http.StatusOK) + }, + ) + + req := httptest.NewRequest(http.MethodGet, "/emails/"+tt.pathID.String(), nil) + rec := httptest.NewRecorder() + r.ServeHTTP(rec, req) + + if rec.Code != tt.wantStatus { + t.Fatalf("status = %d, want %d; body=%s", rec.Code, tt.wantStatus, rec.Body.String()) + } + }) + } +} + func TestRequireAPIPermission(t *testing.T) { tests := []struct { name string diff --git a/internal/api/routes.go b/internal/api/routes.go index 66912edc..6a161b4f 100644 --- a/internal/api/routes.go +++ b/internal/api/routes.go @@ -147,11 +147,11 @@ func Run( emails.Use(m.RateLimitMiddleware(models.RateLimitWrite)) { emails.GET("", m.RequireAccess(models.PermViewCampaigns, models.APIPermReadEmails), h.EmailsSearch) - emails.GET("/:id", m.RequireAccess(models.PermViewCampaigns, models.APIPermReadEmails), h.GetEmail) - emails.PATCH("/:id", m.RequireAccess(models.PermManageEmails, models.APIPermWriteEmails), h.UpdateEmail) - emails.PATCH("/:id/track", m.RequireAccess(models.PermManageEmails, models.APIPermWriteEmails), h.UpdateEmailTrackingDomain) - emails.DELETE("/:id", m.RequireAccess(models.PermManageEmails, models.APIPermWriteEmails), h.DeleteEmail) - emails.POST("/:id/send", m.RequireOrganization(), m.RequireAccess(models.PermSendCampaigns, models.APIPermSendCampaigns), h.SendEmailFromAccount) + emails.GET("/:id", m.RequireAccess(models.PermViewCampaigns, models.APIPermReadEmails), middleware.RequireAPIKeyEmailAccountParam("id"), h.GetEmail) + emails.PATCH("/:id", m.RequireAccess(models.PermManageEmails, models.APIPermWriteEmails), middleware.RequireAPIKeyEmailAccountParam("id"), h.UpdateEmail) + emails.PATCH("/:id/track", m.RequireAccess(models.PermManageEmails, models.APIPermWriteEmails), middleware.RequireAPIKeyEmailAccountParam("id"), h.UpdateEmailTrackingDomain) + emails.DELETE("/:id", m.RequireAccess(models.PermManageEmails, models.APIPermWriteEmails), middleware.RequireAPIKeyEmailAccountParam("id"), h.DeleteEmail) + emails.POST("/:id/send", m.RequireOrganization(), m.RequireAccess(models.PermSendCampaigns, models.APIPermSendCampaigns), middleware.RequireAPIKeyEmailAccountParam("id"), h.SendEmailFromAccount) } // Email onboarding is JWT-only — it writes user-encrypted refresh diff --git a/internal/app/analytics/service.go b/internal/app/analytics/service.go index 96c2a340..68d31bfc 100644 --- a/internal/app/analytics/service.go +++ b/internal/app/analytics/service.go @@ -213,7 +213,7 @@ func (s *analyticsService) GetAccountStatus(ctx context.Context, userID, account func (s *analyticsService) GetAllAccountStatuses(ctx context.Context, userID uuid.UUID) ([]models.EmailAccountStatus, *errx.Error) { // Get all email accounts for user - emailsResult, xerr := s.emailRepo.Search(ctx, userID.String(), "", nil, nil, 1000) + emailsResult, xerr := s.emailRepo.Search(ctx, userID.String(), "", nil, nil, 1000, nil) if xerr != nil { return nil, xerr } diff --git a/internal/app/email/handler.go b/internal/app/email/handler.go index ca442742..40296ee8 100644 --- a/internal/app/email/handler.go +++ b/internal/app/email/handler.go @@ -3,13 +3,14 @@ package email import ( "context" + "github.com/google/uuid" "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/infrastructure/pubsub" "github.com/warmbly/warmbly/internal/models" "github.com/warmbly/warmbly/internal/utils/validate" ) -func (s *emailService) Search(ctx context.Context, userID, search, cursor, tag, limit string) (*models.EmailsResult, *errx.Error) { +func (s *emailService) Search(ctx context.Context, userID, search, cursor, tag, limit string, allowedAccountIDs []uuid.UUID) (*models.EmailsResult, *errx.Error) { cursorId, err := validate.Uuid(cursor) if err != nil { return nil, err @@ -28,7 +29,7 @@ func (s *emailService) Search(ctx context.Context, userID, search, cursor, tag, return nil, err } - return s.emailRepository.Search(ctx, userID, search, cursorId, tagId, limitN) + return s.emailRepository.Search(ctx, userID, search, cursorId, tagId, limitN, allowedAccountIDs) } func (s *emailService) Get(ctx context.Context, userID, emailAccountID string) (*models.Email, *errx.Error) { diff --git a/internal/app/email/service.go b/internal/app/email/service.go index 2c8b3787..6264d833 100644 --- a/internal/app/email/service.go +++ b/internal/app/email/service.go @@ -21,7 +21,7 @@ import ( ) type EmailService interface { - Search(ctx context.Context, userID, search, cursor, tag, limit string) (*models.EmailsResult, *errx.Error) + Search(ctx context.Context, userID, search, cursor, tag, limit string, allowedAccountIDs []uuid.UUID) (*models.EmailsResult, *errx.Error) Get(ctx context.Context, userID, emailAccountID string) (*models.Email, *errx.Error) Update(ctx context.Context, userID, emailAccountID string, udata *models.UpdateEmail) (*models.Email, *errx.Error) UpdateTrackingDomain(ctx context.Context, userID, emailAccountID, domain string) *errx.Error diff --git a/internal/repository/pg_email.go b/internal/repository/pg_email.go index 512d2b21..3e257d1a 100644 --- a/internal/repository/pg_email.go +++ b/internal/repository/pg_email.go @@ -40,7 +40,7 @@ type OAuthCredentials struct { } type EmailRepository interface { - Search(ctx context.Context, userID, search string, cursor, tag *string, limit int32) (*models.EmailsResult, *errx.Error) + Search(ctx context.Context, userID, search string, cursor, tag *string, limit int32, allowedAccountIDs []uuid.UUID) (*models.EmailsResult, *errx.Error) Get(ctx context.Context, userID, emailAccountID string) (*models.Email, *errx.Error) GetByID(ctx context.Context, emailAccountID uuid.UUID) (*models.Email, *errx.Error) GetByTags(ctx context.Context, userID string, tags []string) ([]models.Email, *errx.Error) @@ -350,7 +350,7 @@ func (r *emailRepository) NewSMTPIMAPAccount(ctx context.Context, userID string, }, nil } -func (r *emailRepository) Search(ctx context.Context, userID, search string, cursor, tag *string, limit int32) (*models.EmailsResult, *errx.Error) { +func (r *emailRepository) Search(ctx context.Context, userID, search string, cursor, tag *string, limit int32, allowedAccountIDs []uuid.UUID) (*models.EmailsResult, *errx.Error) { tx, err := r.DB.Begin(ctx) if err != nil { db.CaptureError(err, "", nil, "begin") @@ -384,17 +384,23 @@ func (r *emailRepository) Search(ctx context.Context, userID, search string, cur AND ($4::uuid IS NULL OR EXISTS ( SELECT 1 FROM email_tags cf WHERE cf.email_id = ea.id AND cf.tag_id = $4 )) + AND ($6::uuid[] IS NULL OR ea.id = ANY($6::uuid[])) GROUP BY ea.id ORDER BY ea.created_at DESC, ea.id DESC LIMIT $5 ` + var allowedAccountParam any + if len(allowedAccountIDs) > 0 { + allowedAccountParam = allowedAccountIDs + } params := []any{ userID, cursor, "%" + search + "%", tag, limit + 1, + allowedAccountParam, } rows, err := tx.Query(ctx, query, params...) @@ -441,12 +447,14 @@ func (r *emailRepository) Search(ctx context.Context, userID, search string, cur AND ($3::uuid IS NULL OR EXISTS ( SELECT 1 FROM email_tags cf WHERE cf.email_id = ea.id AND cf.tag_id = $3 )) + AND ($4::uuid[] IS NULL OR ea.id = ANY($4::uuid[])) ` params = []any{ userID, "%" + search + "%", tag, + allowedAccountParam, } var tmp int64 From 640da62b32577b08a1c40bf41df486fea9358449 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 30 May 2026 04:31:43 +0000 Subject: [PATCH 05/11] feat: add api idempotency keys --- cmd/backend/main.go | 4 + internal/api/middleware/handler.go | 2 + internal/api/middleware/idempotency.go | 158 ++++++++++++++++++ internal/api/middleware/idempotency_test.go | 108 ++++++++++++ internal/api/routes.go | 2 +- internal/app/idempotency/service.go | 129 ++++++++++++++ .../000048_api_idempotency_keys.down.sql | 1 + .../000048_api_idempotency_keys.up.sql | 19 +++ 8 files changed, 422 insertions(+), 1 deletion(-) create mode 100644 internal/api/middleware/idempotency.go create mode 100644 internal/api/middleware/idempotency_test.go create mode 100644 internal/app/idempotency/service.go create mode 100644 internal/infrastructure/db/migrations/000048_api_idempotency_keys.down.sql create mode 100644 internal/infrastructure/db/migrations/000048_api_idempotency_keys.up.sql diff --git a/cmd/backend/main.go b/cmd/backend/main.go index 51775e9e..ea24adbb 100644 --- a/cmd/backend/main.go +++ b/cmd/backend/main.go @@ -34,6 +34,7 @@ import ( "github.com/warmbly/warmbly/internal/app/feature" "github.com/warmbly/warmbly/internal/app/fleet" "github.com/warmbly/warmbly/internal/app/group" + idempotencyapp "github.com/warmbly/warmbly/internal/app/idempotency" "github.com/warmbly/warmbly/internal/app/integration" "github.com/warmbly/warmbly/internal/app/organization" "github.com/warmbly/warmbly/internal/app/releases" @@ -111,6 +112,7 @@ func main() { var categoryService group.GroupService var crmService crm.CRMService var apiKeyService apikey.APIKeyService + var idempotencyService idempotencyapp.Service // New services for trial, feature gates, and worker assignment var trialService trial.TrialService @@ -442,6 +444,7 @@ func main() { organizationRepoForHandler = organizationRepository taskRepository := repository.NewTaskRepository(primaryDB.Pool) apiKeyRepository := repository.NewAPIKeyRepository(primaryDB) + idempotencyService = idempotencyapp.NewService(primaryDB.Pool) crmRepository := repository.NewCRMRepository(primaryDB.Pool) advancedRepository := repository.NewAdvancedOutreachRepository(primaryDB.Pool) templateRepository := repository.NewTemplateRepository(primaryDB.Pool) @@ -824,6 +827,7 @@ func main() { m := &middleware.Handler{ TokenService: tokenService, APIKeyService: apiKeyService, + IdempotencyService: idempotencyService, OrganizationService: organizationService, } diff --git a/internal/api/middleware/handler.go b/internal/api/middleware/handler.go index aaff3c87..3f603fc8 100644 --- a/internal/api/middleware/handler.go +++ b/internal/api/middleware/handler.go @@ -2,6 +2,7 @@ package middleware import ( "github.com/warmbly/warmbly/internal/app/apikey" + "github.com/warmbly/warmbly/internal/app/idempotency" "github.com/warmbly/warmbly/internal/app/organization" "github.com/warmbly/warmbly/internal/app/ratelimit" "github.com/warmbly/warmbly/internal/app/token" @@ -10,6 +11,7 @@ import ( type Handler struct { TokenService token.TokenService APIKeyService apikey.APIKeyService + IdempotencyService idempotency.Service RateLimitService ratelimit.RateLimitService OrganizationService organization.OrganizationService } diff --git a/internal/api/middleware/idempotency.go b/internal/api/middleware/idempotency.go new file mode 100644 index 00000000..d569aed9 --- /dev/null +++ b/internal/api/middleware/idempotency.go @@ -0,0 +1,158 @@ +package middleware + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "io" + "net/http" + "strings" + + "github.com/gin-gonic/gin" + "github.com/warmbly/warmbly/internal/app/idempotency" + "github.com/warmbly/warmbly/internal/errx" +) + +const ( + IdempotencyKeyHeader = "Idempotency-Key" + IdempotencyReplayedHeader = "X-Idempotent-Replayed" +) + +// IdempotencyMiddleware implements Stripe-style retry safety for mutating API +// requests. It is opt-in per request via Idempotency-Key and scoped by +// organization, so the same key cannot collide across tenants. +func (h *Handler) IdempotencyMiddleware() gin.HandlerFunc { + return func(c *gin.Context) { + key := strings.TrimSpace(c.GetHeader(IdempotencyKeyHeader)) + if key == "" || !isIdempotentMethod(c.Request.Method) { + c.Next() + return + } + if h.IdempotencyService == nil { + errx.Handle(c, errx.New(errx.ServiceUnavailable, "idempotency service is not available")) + c.Abort() + return + } + if !validIdempotencyKey(key) { + errx.Handle(c, errx.New(errx.BadRequest, "Idempotency-Key must be 1-255 visible ASCII characters")) + c.Abort() + return + } + + orgID := GetOrganizationID(c) + if orgID == nil { + errx.Handle(c, errx.New(errx.BadRequest, "Idempotency-Key requires an organization context")) + c.Abort() + return + } + + body, err := readAndRestoreBody(c) + if err != nil { + errx.Handle(c, errx.New(errx.BadRequest, "failed to read request body")) + c.Abort() + return + } + + path := c.FullPath() + if path == "" { + path = c.Request.URL.Path + } + requestHash := hashRequest(c.Request.Method, path, c.Request.URL.RawQuery, body) + record, state, xerr := h.IdempotencyService.Begin(c.Request.Context(), *orgID, key, c.Request.Method, path, requestHash) + if xerr != nil { + errx.Handle(c, xerr) + c.Abort() + return + } + + switch state { + case idempotency.StateReplay: + c.Header(IdempotencyReplayedHeader, "true") + if record.ContentType != nil && *record.ContentType != "" { + c.Header("Content-Type", *record.ContentType) + } + c.Data(record.StatusCode, c.Writer.Header().Get("Content-Type"), record.ResponseBody) + c.Abort() + return + case idempotency.StateProcessing: + errx.Handle(c, errx.New(errx.Conflict, "an identical request is still processing")) + c.Abort() + return + case idempotency.StateConflict: + errx.Handle(c, errx.New(errx.Conflict, "Idempotency-Key was already used with a different request")) + c.Abort() + return + } + + capture := &captureResponseWriter{ResponseWriter: c.Writer} + c.Writer = capture + c.Next() + + status := capture.Status() + if status == 0 { + status = http.StatusOK + } + contentType := capture.Header().Get("Content-Type") + _ = h.IdempotencyService.Complete(c.Request.Context(), record.ID, status, capture.body.Bytes(), contentType) + } +} + +func isIdempotentMethod(method string) bool { + switch method { + case http.MethodPost, http.MethodPut, http.MethodPatch, http.MethodDelete: + return true + default: + return false + } +} + +func validIdempotencyKey(key string) bool { + if key == "" || len(key) > 255 { + return false + } + for _, r := range key { + if r < 33 || r > 126 { + return false + } + } + return true +} + +func readAndRestoreBody(c *gin.Context) ([]byte, error) { + if c.Request.Body == nil { + return nil, nil + } + body, err := io.ReadAll(c.Request.Body) + if err != nil { + return nil, err + } + c.Request.Body = io.NopCloser(bytes.NewReader(body)) + return body, nil +} + +func hashRequest(method, path, rawQuery string, body []byte) string { + h := sha256.New() + h.Write([]byte(method)) + h.Write([]byte{0}) + h.Write([]byte(path)) + h.Write([]byte{0}) + h.Write([]byte(rawQuery)) + h.Write([]byte{0}) + h.Write(body) + return hex.EncodeToString(h.Sum(nil)) +} + +type captureResponseWriter struct { + gin.ResponseWriter + body bytes.Buffer +} + +func (w *captureResponseWriter) Write(data []byte) (int, error) { + w.body.Write(data) + return w.ResponseWriter.Write(data) +} + +func (w *captureResponseWriter) WriteString(data string) (int, error) { + w.body.WriteString(data) + return w.ResponseWriter.WriteString(data) +} diff --git a/internal/api/middleware/idempotency_test.go b/internal/api/middleware/idempotency_test.go new file mode 100644 index 00000000..abb25e0d --- /dev/null +++ b/internal/api/middleware/idempotency_test.go @@ -0,0 +1,108 @@ +package middleware + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gin-gonic/gin" + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/app/idempotency" + "github.com/warmbly/warmbly/internal/errx" +) + +type fakeIdempotencyService struct { + record *idempotency.Record + state idempotency.State + body []byte +} + +func (s *fakeIdempotencyService) Begin(ctx context.Context, orgID uuid.UUID, key, method, path, requestHash string) (*idempotency.Record, idempotency.State, *errx.Error) { + if s.record == nil { + s.record = &idempotency.Record{ID: uuid.New(), StatusCode: http.StatusCreated} + } + return s.record, s.state, nil +} + +func (s *fakeIdempotencyService) Complete(ctx context.Context, recordID uuid.UUID, statusCode int, responseBody []byte, contentType string) *errx.Error { + s.record.StatusCode = statusCode + s.body = append([]byte(nil), responseBody...) + return nil +} + +func TestIdempotencyMiddlewareStoresResponse(t *testing.T) { + gin.SetMode(gin.TestMode) + orgID := uuid.New() + svc := &fakeIdempotencyService{state: idempotency.StateStarted} + h := &Handler{IdempotencyService: svc} + + r := gin.New() + r.Use(RequestIDMiddleware()) + r.POST("/contacts", + func(c *gin.Context) { + c.Set(OrganizationIDKey, orgID) + c.Next() + }, + h.IdempotencyMiddleware(), + func(c *gin.Context) { + c.JSON(http.StatusCreated, gin.H{"id": "contact_123"}) + }, + ) + + req := httptest.NewRequest(http.MethodPost, "/contacts", nil) + req.Header.Set(IdempotencyKeyHeader, "idem_123") + rec := httptest.NewRecorder() + r.ServeHTTP(rec, req) + + if rec.Code != http.StatusCreated { + t.Fatalf("status = %d, want %d", rec.Code, http.StatusCreated) + } + if string(svc.body) == "" { + t.Fatal("expected response body to be stored") + } +} + +func TestIdempotencyMiddlewareReplaysResponse(t *testing.T) { + gin.SetMode(gin.TestMode) + orgID := uuid.New() + contentType := "application/json; charset=utf-8" + svc := &fakeIdempotencyService{ + state: idempotency.StateReplay, + record: &idempotency.Record{ + ID: uuid.New(), + StatusCode: http.StatusCreated, + ResponseBody: []byte(`{"id":"contact_123"}`), + ContentType: &contentType, + }, + } + h := &Handler{IdempotencyService: svc} + + r := gin.New() + r.Use(RequestIDMiddleware()) + r.POST("/contacts", + func(c *gin.Context) { + c.Set(OrganizationIDKey, orgID) + c.Next() + }, + h.IdempotencyMiddleware(), + func(c *gin.Context) { + c.JSON(http.StatusTeapot, gin.H{"unexpected": true}) + }, + ) + + req := httptest.NewRequest(http.MethodPost, "/contacts", nil) + req.Header.Set(IdempotencyKeyHeader, "idem_123") + rec := httptest.NewRecorder() + r.ServeHTTP(rec, req) + + if rec.Code != http.StatusCreated { + t.Fatalf("status = %d, want %d", rec.Code, http.StatusCreated) + } + if rec.Header().Get(IdempotencyReplayedHeader) != "true" { + t.Fatalf("missing replay header") + } + if got := rec.Body.String(); got != `{"id":"contact_123"}` { + t.Fatalf("body = %q", got) + } +} diff --git a/internal/api/routes.go b/internal/api/routes.go index 6a161b4f..c3574dc7 100644 --- a/internal/api/routes.go +++ b/internal/api/routes.go @@ -141,7 +141,7 @@ func Run( // auth types; APIKeyUsageMiddleware records one log row per API-key // request (JWT requests are skipped). protected := r.Group("") - protected.Use(m.CombinedAuthMiddleware(), m.APIKeyUsageMiddleware()) + protected.Use(m.CombinedAuthMiddleware(), m.APIKeyUsageMiddleware(), m.IdempotencyMiddleware()) { emails := protected.Group("/emails") emails.Use(m.RateLimitMiddleware(models.RateLimitWrite)) diff --git a/internal/app/idempotency/service.go b/internal/app/idempotency/service.go new file mode 100644 index 00000000..0a657ae5 --- /dev/null +++ b/internal/app/idempotency/service.go @@ -0,0 +1,129 @@ +package idempotency + +import ( + "context" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/warmbly/warmbly/internal/errx" +) + +const ttl = 24 * time.Hour + +type State string + +const ( + StateStarted State = "started" + StateReplay State = "replay" + StateProcessing State = "processing" + StateConflict State = "conflict" +) + +type Record struct { + ID uuid.UUID + Method string + Path string + RequestHash string + Status string + StatusCode int + ResponseBody []byte + ContentType *string +} + +type Service interface { + Begin(ctx context.Context, orgID uuid.UUID, key, method, path, requestHash string) (*Record, State, *errx.Error) + Complete(ctx context.Context, recordID uuid.UUID, statusCode int, responseBody []byte, contentType string) *errx.Error +} + +type service struct { + db *pgxpool.Pool +} + +func NewService(db *pgxpool.Pool) Service { + return &service{db: db} +} + +func (s *service) Begin(ctx context.Context, orgID uuid.UUID, key, method, path, requestHash string) (*Record, State, *errx.Error) { + if s == nil || s.db == nil { + return nil, "", errx.New(errx.ServiceUnavailable, "idempotency service is not available") + } + + _, _ = s.db.Exec(ctx, ` + DELETE FROM api_idempotency_keys + WHERE organization_id = $1 AND key = $2 AND expires_at < now() + `, orgID, key) + + var id uuid.UUID + err := s.db.QueryRow(ctx, ` + INSERT INTO api_idempotency_keys ( + organization_id, key, method, path, request_hash, status, expires_at + ) + VALUES ($1, $2, $3, $4, $5, 'processing', now() + ($6::integer * interval '1 second')) + ON CONFLICT (organization_id, key) DO NOTHING + RETURNING id + `, orgID, key, method, path, requestHash, int(ttl.Seconds())).Scan(&id) + if err == nil { + return &Record{ID: id, Method: method, Path: path, RequestHash: requestHash, Status: "processing"}, StateStarted, nil + } + if err != pgx.ErrNoRows { + return nil, "", errx.InternalError() + } + + record, xerr := s.get(ctx, orgID, key) + if xerr != nil { + return nil, "", xerr + } + if record.Method != method || record.Path != path || record.RequestHash != requestHash { + return record, StateConflict, nil + } + if record.Status == "completed" { + return record, StateReplay, nil + } + return record, StateProcessing, nil +} + +func (s *service) Complete(ctx context.Context, recordID uuid.UUID, statusCode int, responseBody []byte, contentType string) *errx.Error { + if s == nil || s.db == nil { + return errx.New(errx.ServiceUnavailable, "idempotency service is not available") + } + _, err := s.db.Exec(ctx, ` + UPDATE api_idempotency_keys + SET status = 'completed', + status_code = $2, + response_body = $3, + content_type = NULLIF($4, ''), + updated_at = now() + WHERE id = $1 + `, recordID, statusCode, responseBody, contentType) + if err != nil { + return errx.InternalError() + } + return nil +} + +func (s *service) get(ctx context.Context, orgID uuid.UUID, key string) (*Record, *errx.Error) { + var record Record + err := s.db.QueryRow(ctx, ` + SELECT id, method, path, request_hash, status, COALESCE(status_code, 0), COALESCE(response_body, ''::bytea), content_type + FROM api_idempotency_keys + WHERE organization_id = $1 AND key = $2 + `, orgID, key).Scan( + &record.ID, + &record.Method, + &record.Path, + &record.RequestHash, + &record.Status, + &record.StatusCode, + &record.ResponseBody, + &record.ContentType, + ) + if err == pgx.ErrNoRows { + return nil, errx.ErrNotFound + } + if err != nil { + return nil, errx.InternalError() + } + return &record, nil +} diff --git a/internal/infrastructure/db/migrations/000048_api_idempotency_keys.down.sql b/internal/infrastructure/db/migrations/000048_api_idempotency_keys.down.sql new file mode 100644 index 00000000..9c344b59 --- /dev/null +++ b/internal/infrastructure/db/migrations/000048_api_idempotency_keys.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS api_idempotency_keys; diff --git a/internal/infrastructure/db/migrations/000048_api_idempotency_keys.up.sql b/internal/infrastructure/db/migrations/000048_api_idempotency_keys.up.sql new file mode 100644 index 00000000..c1fa17af --- /dev/null +++ b/internal/infrastructure/db/migrations/000048_api_idempotency_keys.up.sql @@ -0,0 +1,19 @@ +CREATE TABLE IF NOT EXISTS api_idempotency_keys ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + organization_id UUID NOT NULL REFERENCES organizations(id) ON DELETE CASCADE, + key TEXT NOT NULL, + method TEXT NOT NULL, + path TEXT NOT NULL, + request_hash TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('processing', 'completed')), + status_code INTEGER, + response_body BYTEA, + content_type TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + expires_at TIMESTAMPTZ NOT NULL, + UNIQUE (organization_id, key) +); + +CREATE INDEX IF NOT EXISTS idx_api_idempotency_keys_expires + ON api_idempotency_keys (expires_at); From cb0c2b4fac4a588e42226880e85d818795123780 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 30 May 2026 04:33:43 +0000 Subject: [PATCH 06/11] feat: harden webhook endpoints --- internal/api/handler/warmup_routing.go | 2 +- internal/api/handler/webhook.go | 32 +++++++++++-------- internal/app/webhook/service.go | 31 +++++++++++++++--- internal/app/webhook/service_test.go | 13 +++++++- .../000049_webhook_delivery_safety.down.sql | 7 ++++ .../000049_webhook_delivery_safety.up.sql | 14 ++++++++ 6 files changed, 79 insertions(+), 20 deletions(-) create mode 100644 internal/infrastructure/db/migrations/000049_webhook_delivery_safety.down.sql create mode 100644 internal/infrastructure/db/migrations/000049_webhook_delivery_safety.up.sql diff --git a/internal/api/handler/warmup_routing.go b/internal/api/handler/warmup_routing.go index 449fd8b1..64542742 100644 --- a/internal/api/handler/warmup_routing.go +++ b/internal/api/handler/warmup_routing.go @@ -164,6 +164,6 @@ func requireOrgID(c *gin.Context) (uuid.UUID, bool) { if orgID := middleware.GetOrganizationID(c); orgID != nil { return *orgID, true } - c.JSON(http.StatusForbidden, gin.H{"error": "organization context required"}) + errx.JSON(c, errx.New(errx.Forbidden, "organization context required")) return uuid.Nil, false } diff --git a/internal/api/handler/webhook.go b/internal/api/handler/webhook.go index 0206094d..f4f42ad0 100644 --- a/internal/api/handler/webhook.go +++ b/internal/api/handler/webhook.go @@ -7,6 +7,7 @@ import ( "github.com/gin-gonic/gin" "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/errx" "github.com/warmbly/warmbly/internal/models" ) @@ -29,7 +30,7 @@ func (h *Handler) ListWebhookEndpoints(c *gin.Context) { } endpoints, err := h.WebhookService.ListEndpoints(c.Request.Context(), orgID) if err != nil { - c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list endpoints"}) + errx.JSON(c, errx.New(errx.Internal, "failed to list endpoints")) return } if endpoints == nil { @@ -51,7 +52,7 @@ func (h *Handler) CreateWebhookEndpoint(c *gin.Context) { } var p webhookEndpointPayload if err := c.ShouldBindJSON(&p); err != nil { - c.JSON(http.StatusBadRequest, gin.H{"error": "invalid payload"}) + errx.JSON(c, errx.New(errx.BadRequest, "invalid payload")) return } enabled := true @@ -60,7 +61,7 @@ func (h *Handler) CreateWebhookEndpoint(c *gin.Context) { } endpoint, err := h.WebhookService.CreateEndpoint(c.Request.Context(), orgID, p.URL, p.Description, p.EventTypes, enabled) if err != nil { - c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + errx.JSON(c, errx.New(errx.BadRequest, err.Error())) return } c.JSON(http.StatusCreated, endpoint) @@ -75,12 +76,12 @@ func (h *Handler) UpdateWebhookEndpoint(c *gin.Context) { } endpointID, err := uuid.Parse(c.Param("id")) if err != nil { - c.JSON(http.StatusBadRequest, gin.H{"error": "invalid endpoint id"}) + errx.JSON(c, errx.New(errx.BadRequest, "invalid endpoint id")) return } var p webhookEndpointPayload if err := c.ShouldBindJSON(&p); err != nil { - c.JSON(http.StatusBadRequest, gin.H{"error": "invalid payload"}) + errx.JSON(c, errx.New(errx.BadRequest, "invalid payload")) return } enabled := true @@ -89,7 +90,7 @@ func (h *Handler) UpdateWebhookEndpoint(c *gin.Context) { } endpoint, err := h.WebhookService.UpdateEndpoint(c.Request.Context(), orgID, endpointID, p.URL, p.Description, p.EventTypes, enabled) if err != nil { - c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) + errx.JSON(c, errx.New(errx.BadRequest, err.Error())) return } c.JSON(http.StatusOK, endpoint) @@ -104,11 +105,11 @@ func (h *Handler) DeleteWebhookEndpoint(c *gin.Context) { } endpointID, err := uuid.Parse(c.Param("id")) if err != nil { - c.JSON(http.StatusBadRequest, gin.H{"error": "invalid endpoint id"}) + errx.JSON(c, errx.New(errx.BadRequest, "invalid endpoint id")) return } if err := h.WebhookService.DeleteEndpoint(c.Request.Context(), orgID, endpointID); err != nil { - c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) + errx.JSON(c, errx.New(errx.NotFound, err.Error())) return } c.Status(http.StatusNoContent) @@ -124,12 +125,12 @@ func (h *Handler) RotateWebhookSecret(c *gin.Context) { } endpointID, err := uuid.Parse(c.Param("id")) if err != nil { - c.JSON(http.StatusBadRequest, gin.H{"error": "invalid endpoint id"}) + errx.JSON(c, errx.New(errx.BadRequest, "invalid endpoint id")) return } secret, err := h.WebhookService.RotateSecret(c.Request.Context(), orgID, endpointID) if err != nil { - c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) + errx.JSON(c, errx.New(errx.NotFound, err.Error())) return } c.JSON(http.StatusOK, gin.H{"secret": secret}) @@ -144,18 +145,21 @@ func (h *Handler) ListWebhookDeliveries(c *gin.Context) { } endpointID, err := uuid.Parse(c.Param("id")) if err != nil { - c.JSON(http.StatusBadRequest, gin.H{"error": "invalid endpoint id"}) + errx.JSON(c, errx.New(errx.BadRequest, "invalid endpoint id")) return } limit := 50 if raw := c.Query("limit"); raw != "" { - if n, err := strconv.Atoi(raw); err == nil && n > 0 { - limit = n + n, err := strconv.Atoi(raw) + if err != nil || n <= 0 || n > 200 { + errx.JSON(c, errx.New(errx.BadRequest, "limit must be between 1 and 200")) + return } + limit = n } deliveries, err := h.WebhookService.ListDeliveries(c.Request.Context(), orgID, endpointID, limit) if err != nil { - c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list deliveries"}) + errx.JSON(c, errx.New(errx.Internal, "failed to list deliveries")) return } if deliveries == nil { diff --git a/internal/app/webhook/service.go b/internal/app/webhook/service.go index 68fa087d..81ae4b3d 100644 --- a/internal/app/webhook/service.go +++ b/internal/app/webhook/service.go @@ -14,8 +14,10 @@ import ( "encoding/json" "fmt" "io" + "net" "net/http" "net/url" + "os" "strings" "time" @@ -186,8 +188,9 @@ func generateSecret() (string, error) { } // validateURL keeps malformed entries and obvious SSRF targets out of the -// table. We do not enforce HTTPS at insert time because internal-network -// integrations and ngrok-style local tests legitimately use http://. +// table. Public webhook endpoints must use HTTPS and route to public hosts. +// Local/self-hosted development can set WARMBLY_ALLOW_UNSAFE_WEBHOOK_URLS=true +// to permit HTTP and private targets. func validateURL(raw string) error { raw = strings.TrimSpace(raw) if raw == "" { @@ -197,15 +200,35 @@ func validateURL(raw string) error { if err != nil { return fmt.Errorf("invalid url: %w", err) } - if u.Scheme != "http" && u.Scheme != "https" { - return fmt.Errorf("url scheme must be http or https") + allowUnsafe := strings.EqualFold(os.Getenv("WARMBLY_ALLOW_UNSAFE_WEBHOOK_URLS"), "true") + if u.Scheme != "https" && !(allowUnsafe && u.Scheme == "http") { + return fmt.Errorf("url scheme must be https") } if u.Host == "" { return fmt.Errorf("url must have a host") } + if !allowUnsafe && isPrivateWebhookHost(u.Hostname()) { + return fmt.Errorf("url host must be publicly routable") + } return nil } +func isPrivateWebhookHost(host string) bool { + host = strings.Trim(strings.ToLower(host), "[]") + if host == "" || host == "localhost" || strings.HasSuffix(host, ".localhost") { + return true + } + ip := net.ParseIP(host) + if ip == nil { + return false + } + return ip.IsLoopback() || + ip.IsPrivate() || + ip.IsLinkLocalUnicast() || + ip.IsLinkLocalMulticast() || + ip.IsUnspecified() +} + func validateEventTypes(eventTypes []string) error { for _, t := range eventTypes { if !models.IsValidWebhookEventType(t) { diff --git a/internal/app/webhook/service_test.go b/internal/app/webhook/service_test.go index 517353c9..43703f4e 100644 --- a/internal/app/webhook/service_test.go +++ b/internal/app/webhook/service_test.go @@ -73,7 +73,10 @@ func TestValidateURL_RejectsBadInput(t *testing.T) { "javascript:alert(1)": true, "http://": true, "https://example.com/hook": false, - "http://localhost:3000/hook": false, + "http://localhost:3000/hook": true, + "https://localhost/hook": true, + "https://127.0.0.1/hook": true, + "https://10.0.0.10/hook": true, } for input, wantErr := range cases { err := validateURL(input) @@ -85,3 +88,11 @@ func TestValidateURL_RejectsBadInput(t *testing.T) { } } } + +func TestValidateURL_AllowsUnsafeLocalDevelopmentWhenEnabled(t *testing.T) { + t.Setenv("WARMBLY_ALLOW_UNSAFE_WEBHOOK_URLS", "true") + + if err := validateURL("http://localhost:3000/hook"); err != nil { + t.Fatalf("expected unsafe local URL to be accepted in development mode: %v", err) + } +} diff --git a/internal/infrastructure/db/migrations/000049_webhook_delivery_safety.down.sql b/internal/infrastructure/db/migrations/000049_webhook_delivery_safety.down.sql new file mode 100644 index 00000000..2e2f4a9c --- /dev/null +++ b/internal/infrastructure/db/migrations/000049_webhook_delivery_safety.down.sql @@ -0,0 +1,7 @@ +DROP INDEX IF EXISTS idx_webhook_deliveries_endpoint_event; + +DROP INDEX IF EXISTS idx_webhook_deliveries_due; + +CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_due + ON webhook_deliveries (next_attempt_at) + WHERE status IN ('pending', 'retry'); diff --git a/internal/infrastructure/db/migrations/000049_webhook_delivery_safety.up.sql b/internal/infrastructure/db/migrations/000049_webhook_delivery_safety.up.sql new file mode 100644 index 00000000..f9e0beba --- /dev/null +++ b/internal/infrastructure/db/migrations/000049_webhook_delivery_safety.up.sql @@ -0,0 +1,14 @@ +DROP INDEX IF EXISTS idx_webhook_deliveries_due; + +CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_due + ON webhook_deliveries (next_attempt_at) + WHERE status = 'pending'; + +DELETE FROM webhook_deliveries a +USING webhook_deliveries b +WHERE a.endpoint_id = b.endpoint_id + AND a.event_id = b.event_id + AND a.ctid < b.ctid; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_webhook_deliveries_endpoint_event + ON webhook_deliveries (endpoint_id, event_id); From d28c60701363feb8ba73359e8ac9774b120cfe25 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 30 May 2026 04:35:21 +0000 Subject: [PATCH 07/11] feat: document api release standards --- AGENTS.md | 9 +++++++++ docs/content/docs/api-keys/permissions.mdx | 8 +++++--- docs/content/docs/authentication.mdx | 2 ++ docs/content/docs/reference/endpoints.mdx | 9 +++++++++ docs/content/docs/reference/error-codes.mdx | 10 ++++++++-- docs/content/docs/reference/permissions.mdx | 16 +++++++++------- 6 files changed, 42 insertions(+), 12 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 646a02d1..d81a1ee6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -38,6 +38,15 @@ Dashboard realtime: - aim for a responsive, Discord-like product feel: presence, counts, lists, detail panes, notifications, and workflow state should stay current across every dashboard feature where live updates are meaningful - when changing dashboard behavior, it is acceptable to safely change the API structure if a better solution exists. Before making an API shape change, ask the user how they want to handle it, especially when the current API may already be published or backwards compatibility might require a new API version +Public API quality bar: + +- treat customer-facing API changes as contract changes. Prefer additive changes inside a version, and use a new API version for incompatible behavior once an endpoint is published +- every API-key-capable route must have an explicit API permission gate and, for JWT callers, the matching organization permission gate +- side-effectful POST/PATCH/PUT/DELETE endpoints should support `Idempotency-Key` or have a documented reason why retries are naturally safe +- error responses should include stable machine-readable `code` and `request_id` fields in addition to human-readable text +- list endpoints should use consistent `data` plus `pagination` shapes with opaque cursors; invalid cursors or limits should return `400` instead of being ignored +- webhook endpoints must stay HMAC-signed, HTTPS by default, and protected against obvious SSRF targets. Only development/self-hosted environments should opt into unsafe webhook URLs + ## System Shape - `cmd/backend`: API and business orchestration diff --git a/docs/content/docs/api-keys/permissions.mdx b/docs/content/docs/api-keys/permissions.mdx index 9f8e7845..0bcfc4b0 100644 --- a/docs/content/docs/api-keys/permissions.mdx +++ b/docs/content/docs/api-keys/permissions.mdx @@ -52,11 +52,13 @@ curl -X GET "https://api.warmbly.com/api-keys/permissions" \ { "name": "BULK_CAMPAIGNS", "value": 1024, "description": "Bulk campaign operations", "category": "bulk" }, { "name": "REALTIME_SUBSCRIBE", "value": 2048, "description": "Subscribe to realtime events", "category": "special" }, { "name": "WEBHOOKS", "value": 4096, "description": "Manage webhook endpoints", "category": "special" }, - { "name": "API_KEYS", "value": 8192, "description": "Create and manage API keys", "category": "special" } + { "name": "API_KEYS", "value": 8192, "description": "Create and manage API keys", "category": "special" }, + { "name": "INTEGRATIONS", "value": 1048576, "description": "Connect and manage third-party integrations", "category": "special" }, + { "name": "WARMUP_ROUTING", "value": 2097152, "description": "Manage warmup routing rules", "category": "special" } ], "presets": { "read_only": 688159, - "full_access": 1048575 + "full_access": 4194303 } } ``` @@ -75,7 +77,7 @@ curl -X GET "https://api.warmbly.com/api-keys/permissions" \ | Preset | Value | Description | |--------|-------|-------------| | `read_only` | 688159 | All read permissions across emails, campaigns, contacts, unibox, analytics, templates, CRM, audit logs | -| `full_access` | 1048575 | Every defined permission bit | +| `full_access` | 4194303 | Every defined permission bit | ## Working with Permissions diff --git a/docs/content/docs/authentication.mdx b/docs/content/docs/authentication.mdx index d40e3993..50ab915f 100644 --- a/docs/content/docs/authentication.mdx +++ b/docs/content/docs/authentication.mdx @@ -29,6 +29,8 @@ curl -X GET "https://api.warmbly.com/api-keys" \ -H "Content-Type: application/json" ``` +For mutation retries, include an `Idempotency-Key` header with a unique value per logical operation. Warmbly stores completed mutation responses for 24 hours per organization and key, then replays matching retries instead of performing the operation again. + ## Key Security Best Practices diff --git a/docs/content/docs/reference/endpoints.mdx b/docs/content/docs/reference/endpoints.mdx index a8893a7d..497edcad 100644 --- a/docs/content/docs/reference/endpoints.mdx +++ b/docs/content/docs/reference/endpoints.mdx @@ -116,6 +116,15 @@ When an endpoint says "JWT permission: X / API permission: Y", the dual-auth mid | POST | `/deliverability/events` | `WRITE_CAMPAIGNS` | | GET | `/tasks/dlq` | `SEND_CAMPAIGNS` | | POST | `/tasks/dlq/:id/replay` | `SEND_CAMPAIGNS` | +| GET/POST/PATCH/DELETE | `/webhooks[/:id]` | `WEBHOOKS` | +| POST | `/webhooks/:id/rotate-secret` | `WEBHOOKS` | +| GET | `/webhooks/:id/deliveries` | `WEBHOOKS` | +| GET/POST/DELETE | `/integrations/*` | `INTEGRATIONS` | +| GET/POST/PATCH/DELETE | `/warmup/routing[/:id]` | `WARMUP_ROUTING` | + +### Retry safety + +Mutating API requests may include an `Idempotency-Key` header. Warmbly stores the completed response for 24 hours per organization and key. Reusing the same key with the same method, route, query, and body replays the original response with `X-Idempotent-Replayed: true`; reusing the key with a different request returns `409 Conflict`. ### Reference data diff --git a/docs/content/docs/reference/error-codes.mdx b/docs/content/docs/reference/error-codes.mdx index c7e5f4b7..c7adc2b6 100644 --- a/docs/content/docs/reference/error-codes.mdx +++ b/docs/content/docs/reference/error-codes.mdx @@ -14,10 +14,14 @@ All errors follow this structure: ```json { "error": "Error Type", - "message": "Human-readable description of what went wrong." + "message": "Human-readable description of what went wrong.", + "code": "machine_readable_code", + "request_id": "req_or_uuid_for_support" } ``` +`error` and `message` are for people. Client logic should use `code`, HTTP status, and endpoint-specific fields such as `retry_after`. Include `request_id` when contacting support. + ## HTTP Status Codes ### Client Errors (4xx) @@ -56,7 +60,9 @@ Returned when the request cannot be processed due to invalid syntax. ```json { "error": "Bad Request", - "message": "invalid request body" + "message": "invalid request body", + "code": "bad_request", + "request_id": "4bbbd1b2-8f86-47dd-8a7f-9476501ad20e" } ``` diff --git a/docs/content/docs/reference/permissions.mdx b/docs/content/docs/reference/permissions.mdx index c7a51cdf..7865e598 100644 --- a/docs/content/docs/reference/permissions.mdx +++ b/docs/content/docs/reference/permissions.mdx @@ -1,6 +1,6 @@ --- title: Permissions Reference -description: Complete reference for the 20 API permissions available in Warmbly. +description: Complete reference for the 22 API permissions available in Warmbly. --- # Permissions Reference @@ -23,7 +23,7 @@ Warmbly uses a bitmask system for API permissions. Each permission is one bit in | `BULK_CONTACTS` | 9 | 512 | bulk | Bulk import/export/delete contacts | | `BULK_CAMPAIGNS` | 10 | 1024 | bulk | Bulk campaign operations | | `REALTIME_SUBSCRIBE` | 11 | 2048 | special | Subscribe to realtime events | -| `WEBHOOKS` | 12 | 4096 | special | Manage webhook endpoints (reserved) | +| `WEBHOOKS` | 12 | 4096 | special | Manage webhook endpoints | | `API_KEYS` | 13 | 8192 | special | Create and manage API keys | | `SEND_CAMPAIGNS` | 14 | 16384 | write | Start and stop campaigns (sends real mail) | | `READ_TEMPLATES` | 15 | 32768 | read | View reply templates | @@ -31,6 +31,8 @@ Warmbly uses a bitmask system for API permissions. Each permission is one bit in | `READ_CRM` | 17 | 131072 | read | View pipelines, deals, and CRM tasks | | `WRITE_CRM` | 18 | 262144 | write | Create and modify pipelines, deals, CRM tasks | | `READ_AUDIT_LOGS` | 19 | 524288 | read | View organization audit logs | +| `INTEGRATIONS` | 20 | 1048576 | special | Connect and manage third-party integrations | +| `WARMUP_ROUTING` | 21 | 2097152 | special | Manage warmup routing rules | `SEND_CAMPAIGNS` is intentionally separate from `WRITE_CAMPAIGNS`: editing a campaign draft and starting one that actually transmits mail are different blast radii, so a key can be granted the first without the second. @@ -50,7 +52,7 @@ High-volume operations. These can touch large numbers of rows in a single reques ### Special -Realtime subscriptions, webhooks, self-service key management. Grant individually. +Realtime subscriptions, webhooks, integrations, warmup routing, and self-service key management. Grant individually. ## Preset Combinations @@ -67,12 +69,12 @@ READ_EMAILS | READ_CAMPAIGNS | READ_CONTACTS | READ_UNIBOX | READ_ANALYTICS = 688159 ``` -### Full Access — 1048575 +### Full Access — 4194303 -All 20 permissions: +All 22 permissions: ``` -(1 << 20) - 1 = 1048575 +(1 << 22) - 1 = 4194303 ``` ## Working with Bitmasks @@ -101,7 +103,7 @@ hasPermission(16387, 64); // false — WRITE_CAMPAIGNS missing ### Rejecting unknown bits -`POST /api-keys` rejects any request whose `permissions` field has bits outside the known set, so a stale client can't accidentally request a future permission. The current mask of valid bits is `1048575`. +`POST /api-keys` rejects any request whose `permissions` field has bits outside the known set, so a stale client can't accidentally request a future permission. The current mask of valid bits is `4194303`. ## Common Permission Sets From d8a546eca58d1f282c7efc44c9ecc3bcfe206e1a Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 30 May 2026 05:10:27 +0000 Subject: [PATCH 08/11] feat: add worker enrollment install --- cmd/backend/main.go | 29 ++++--- cmd/worker/main.go | 59 +++++++++++++ internal/api/handler/admin_workers_ssh.go | 2 +- internal/api/handler/worker_enrollment.go | 87 +++++++++++++++++++ internal/api/routes.go | 5 ++ .../app/worker_orchestrator/orchestrator.go | 33 +++++++ internal/repository/pg_worker_heartbeat.go | 7 ++ internal/repository/pg_worker_ssh.go | 6 +- scripts/install-worker.sh | 69 ++++++++++++++- 9 files changed, 281 insertions(+), 16 deletions(-) create mode 100644 internal/api/handler/worker_enrollment.go diff --git a/cmd/backend/main.go b/cmd/backend/main.go index ea24adbb..6c468dba 100644 --- a/cmd/backend/main.go +++ b/cmd/backend/main.go @@ -581,18 +581,23 @@ func main() { credentialsRepository, cipherService, worker_orchestrator.WorkerEnvConfig{ - AppEnv: os.Getenv("APP_ENV"), - WorkerImage: getenvDefault("WORKER_IMAGE", "ghcr.io/warmbly/worker:latest"), - KafkaBootstrap: os.Getenv("KAFKA_BOOTSTRAP_SERVERS"), - KafkaSASLUsername: os.Getenv("KAFKA_SASL_USERNAME"), - KafkaSASLPassword: os.Getenv("KAFKA_SASL_PASSWORD"), - SchemaRegistryURL: os.Getenv("SCHEMA_REGISTRY_URL"), - SchemaRegistryKey: os.Getenv("SCHEMA_REGISTRY_KEY"), - SchemaRegistrySecret: os.Getenv("SCHEMA_REGISTRY_SECRET"), - RedisURL: os.Getenv("REDIS"), - AWSRegion: os.Getenv("AWS_REGION"), - AWSAccessKeyID: os.Getenv("WORKER_AWS_ACCESS_KEY_ID"), - AWSSecretAccessKey: os.Getenv("WORKER_AWS_SECRET_ACCESS_KEY"), + AppEnv: os.Getenv("APP_ENV"), + WorkerImage: getenvDefault("WORKER_IMAGE", "ghcr.io/warmbly/worker:latest"), + KafkaBootstrap: os.Getenv("KAFKA_BOOTSTRAP_SERVERS"), + KafkaSASLUsername: os.Getenv("KAFKA_SASL_USERNAME"), + KafkaSASLPassword: os.Getenv("KAFKA_SASL_PASSWORD"), + SchemaRegistryURL: os.Getenv("SCHEMA_REGISTRY_URL"), + SchemaRegistryKey: os.Getenv("SCHEMA_REGISTRY_KEY"), + SchemaRegistrySecret: os.Getenv("SCHEMA_REGISTRY_SECRET"), + RedisURL: os.Getenv("REDIS"), + AWSRegion: os.Getenv("AWS_REGION"), + AWSAccessKeyID: os.Getenv("WORKER_AWS_ACCESS_KEY_ID"), + AWSSecretAccessKey: os.Getenv("WORKER_AWS_SECRET_ACCESS_KEY"), + EncryptedKeysBackendURL: os.Getenv("ENCRYPTED_KEYS_BACKEND_URL"), + EncryptedKeysWorkerToken: os.Getenv("INTERNAL_API_TOKEN"), + EventBusProvider: os.Getenv("EVENTBUS_PROVIDER"), + NATSURL: os.Getenv("NATS_URL"), + CodecProvider: os.Getenv("CODEC_PROVIDER"), }, getenvDefault("WORKER_INSTALLER_PATH", "/app/scripts/install-worker.sh"), ) diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 0b4929e5..5dd33f27 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -1,10 +1,14 @@ package main import ( + "bytes" "context" + "encoding/json" "log" + "net/http" "os" "os/signal" + "strings" "syscall" "time" @@ -164,6 +168,7 @@ func main() { // the rolling 1m counters into a WorkerHealth event, publishes via the // event bus so the consumer can write a row into worker_health_samples. go workerService.Heartbeat(ctx) + go runInternalHeartbeat(ctx, workerID, bindIP) go workerService.RunHealth(ctx, 30*time.Second) // Graceful shutdown @@ -183,6 +188,60 @@ func main() { log.Println("Worker stopped") } +func runInternalHeartbeat(ctx context.Context, workerID uuid.UUID, bindIP string) { + baseURL := strings.TrimRight(os.Getenv("ENCRYPTED_KEYS_BACKEND_URL"), "/") + token := os.Getenv("ENCRYPTED_KEYS_WORKER_TOKEN") + if baseURL == "" || token == "" { + return + } + reportedIP := os.Getenv("WORKER_PUBLIC_IP") + if reportedIP == "" && bindIP != "default route" { + reportedIP = bindIP + } + if reportedIP == "" { + reportedIP = "unknown" + } + + client := &http.Client{Timeout: 10 * time.Second} + send := func() { + payload := map[string]string{ + "worker_id": workerID.String(), + "bind_ip": reportedIP, + "tier": os.Getenv("WORKER_TIER"), + "egress_kind": os.Getenv("WORKER_EGRESS_KIND"), + } + body, _ := json.Marshal(payload) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/api/v1/internal/worker/heartbeat", bytes.NewReader(body)) + if err != nil { + log.Println("failed to build internal heartbeat:", err) + return + } + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + resp, err := client.Do(req) + if err != nil { + log.Println("failed internal heartbeat:", err) + return + } + defer resp.Body.Close() + if resp.StatusCode >= 300 { + log.Println("internal heartbeat returned status", resp.StatusCode) + } + } + + send() + ticker := time.NewTicker(90 * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + send() + } + } +} + // uuidNamespaceURL is the RFC 4122 URL namespace, matching the value used by // scripts/install-worker.sh when deriving the per-IP worker ID. Keep these in // sync: the installer and the worker must agree on the derivation. diff --git a/internal/api/handler/admin_workers_ssh.go b/internal/api/handler/admin_workers_ssh.go index 0be9f65a..a4c45373 100644 --- a/internal/api/handler/admin_workers_ssh.go +++ b/internal/api/handler/admin_workers_ssh.go @@ -107,7 +107,7 @@ func (h *Handler) AdminCreateWorker(c *gin.Context) { errx.JSON(c, errx.New(errx.Internal, "failed to generate enrollment token")) return } - enrollToken = hex.EncodeToString(raw) + enrollToken = "wmenroll_" + hex.EncodeToString(raw) sum := sha256.Sum256([]byte(enrollToken)) enrollHash = hex.EncodeToString(sum[:]) exp := time.Now().Add(2 * time.Hour) diff --git a/internal/api/handler/worker_enrollment.go b/internal/api/handler/worker_enrollment.go new file mode 100644 index 00000000..17622591 --- /dev/null +++ b/internal/api/handler/worker_enrollment.go @@ -0,0 +1,87 @@ +package handler + +import ( + "crypto/sha256" + "encoding/hex" + "net/http" + "strings" + + "github.com/gin-gonic/gin" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" +) + +type workerEnrollmentRequest struct { + Token string `json:"token"` + PublicIP string `json:"public_ip,omitempty"` +} + +// EnrollWorker exchanges a one-time enrollment token for a complete worker +// dotenv file. It is intentionally public: the high-entropy one-time token is +// the credential, and it is consumed atomically before secrets are returned. +func (h *Handler) EnrollWorker(c *gin.Context) { + if h.WorkerRepo == nil || h.WorkerOrchestrator == nil { + errx.JSON(c, errx.New(errx.ServiceUnavailable, "worker enrollment is not configured")) + return + } + + var req workerEnrollmentRequest + if err := c.ShouldBindJSON(&req); err != nil { + errx.JSON(c, errx.New(errx.BadRequest, "invalid request body")) + return + } + req.Token = strings.TrimSpace(req.Token) + if req.Token == "" { + errx.JSON(c, errx.New(errx.BadRequest, "enrollment token is required")) + return + } + + sum := sha256.Sum256([]byte(req.Token)) + worker, err := h.WorkerRepo.ConsumeEnrollmentToken(c.Request.Context(), hex.EncodeToString(sum[:])) + if err != nil { + errx.JSON(c, errx.New(errx.Internal, "failed to consume enrollment token")) + return + } + if worker == nil { + errx.JSON(c, errx.New(errx.Unauthorized, "enrollment token is invalid or expired")) + return + } + + ip := strings.TrimSpace(req.PublicIP) + if ip == "" { + ip = c.ClientIP() + } + if ip != "" { + _ = h.WorkerRepo.RecordEnrolledIP(c.Request.Context(), worker.ID, ip) + } + + envFile, _, err := h.WorkerOrchestrator.RenderEnrollmentEnv(c.Request.Context(), worker.ID) + if err != nil { + errx.JSON(c, errx.New(errx.Internal, "failed to render worker config")) + return + } + envFile += "WORKER_TIER=" + workerTierLabel(worker) + "\n" + if ip != "" { + envFile += "WORKER_PUBLIC_IP=" + ip + "\n" + } + if worker.EgressKind != "" { + envFile += "WORKER_EGRESS_KIND=" + string(worker.EgressKind) + "\n" + } + + c.Header("Content-Type", "text/plain; charset=utf-8") + c.Header("Cache-Control", "no-store") + c.String(http.StatusOK, envFile) +} + +func workerTierLabel(w *models.Worker) string { + if w == nil { + return "shared_premium" + } + if w.WorkerType == models.WorkerTypeDedicated { + return "dedicated" + } + if w.FreeTier { + return "shared_free" + } + return "shared_premium" +} diff --git a/internal/api/routes.go b/internal/api/routes.go index c3574dc7..d02db4e6 100644 --- a/internal/api/routes.go +++ b/internal/api/routes.go @@ -38,6 +38,11 @@ func Run( r.POST("/api/v1/integrations/inbound/calendly/:secret", h.InboundCalendly) r.POST("/api/v1/integrations/inbound/cal-com/:secret", h.InboundCalCom) + // Public worker enrollment. The one-time enrollment token is the + // credential; successful exchange returns a dotenv file for the installer + // and consumes the token. + r.POST("/api/v1/workers/enroll", h.EnrollWorker) + // Public OAuth-bouncer pages used by the mailbox onboarding popup. // The provider redirects here; the page postMessages the code/state // back to the SPA opener which then calls /emails/onboarding/oauth/finish. diff --git a/internal/app/worker_orchestrator/orchestrator.go b/internal/app/worker_orchestrator/orchestrator.go index b4c40f1e..0a33e3e8 100644 --- a/internal/app/worker_orchestrator/orchestrator.go +++ b/internal/app/worker_orchestrator/orchestrator.go @@ -57,6 +57,13 @@ type WorkerEnvConfig struct { AWSRegion string AWSAccessKeyID string AWSSecretAccessKey string + + EncryptedKeysBackendURL string + EncryptedKeysWorkerToken string + + EventBusProvider string + NATSURL string + CodecProvider string } type Orchestrator struct { @@ -143,6 +150,26 @@ func (o *Orchestrator) Install(ctx context.Context, workerID uuid.UUID) error { return o.repo.UpdateInstallState(ctx, workerID, models.WorkerInstallStateInstalled, "") } +// RenderEnrollmentEnv returns a complete dotenv payload for the one-command +// enrollment flow. The token exchange authenticates the caller; this method +// only renders the config the installer writes to disk. +func (o *Orchestrator) RenderEnrollmentEnv(ctx context.Context, workerID uuid.UUID) (string, string, error) { + envContent, image, err := o.renderEnvFile(ctx, workerID) + if err != nil { + return "", "", err + } + var b strings.Builder + b.WriteString("# Warmbly worker enrollment config\n") + b.WriteString("WORKER_ID=") + b.WriteString(workerID.String()) + b.WriteString("\n") + b.WriteString("WARMBLY_WORKER_IMAGE=") + b.WriteString(image) + b.WriteString("\n") + b.WriteString(envContent) + return b.String(), image, nil +} + // ApplyConfig re-writes /etc/warmbly/worker.env from the worker's current // profile + AWS creds and restarts the service. Cheaper than Install — does // not touch Docker or the installer script. Use after a credential change. @@ -578,6 +605,9 @@ func (o *Orchestrator) renderEnvFile(ctx context.Context, workerID uuid.UUID) (e write("AWS_REGION", env.AWSRegion) write("AWS_ACCESS_KEY_ID", env.AWSAccessKeyID) write("AWS_SECRET_ACCESS_KEY", env.AWSSecretAccessKey) + write("ENCRYPTED_KEYS_PROVIDER", "http") + write("ENCRYPTED_KEYS_BACKEND_URL", env.EncryptedKeysBackendURL) + write("ENCRYPTED_KEYS_WORKER_TOKEN", env.EncryptedKeysWorkerToken) write("KAFKA_BOOTSTRAP_SERVERS", env.KafkaBootstrap) write("KAFKA_SASL_USERNAME", env.KafkaSASLUsername) write("KAFKA_SASL_PASSWORD", env.KafkaSASLPassword) @@ -585,6 +615,9 @@ func (o *Orchestrator) renderEnvFile(ctx context.Context, workerID uuid.UUID) (e write("SCHEMA_REGISTRY_KEY", env.SchemaRegistryKey) write("SCHEMA_REGISTRY_SECRET", env.SchemaRegistrySecret) write("REDIS", env.RedisURL) + write("EVENTBUS_PROVIDER", env.EventBusProvider) + write("NATS_URL", env.NATSURL) + write("CODEC_PROVIDER", env.CodecProvider) return b.String(), image, nil } diff --git a/internal/repository/pg_worker_heartbeat.go b/internal/repository/pg_worker_heartbeat.go index 9aa12c45..889be2d0 100644 --- a/internal/repository/pg_worker_heartbeat.go +++ b/internal/repository/pg_worker_heartbeat.go @@ -24,6 +24,13 @@ func (r *workerRepository) UpsertOnHeartbeat(ctx context.Context, id uuid.UUID, VALUES ($1, $2, $3, TRUE, $4, $5, $6, 'healthy', 0) ON CONFLICT (id) DO UPDATE SET ip_addr = EXCLUDED.ip_addr, + active = TRUE, + install_state = CASE + WHEN workers.install_state IN ('pending', 'provisioning', 'error') THEN 'installed'::worker_install_state + ELSE workers.install_state + END, + last_seen_at = now(), + last_error = NULL, updated_at = now() ` name := "auto-registered-" + id.String()[:8] diff --git a/internal/repository/pg_worker_ssh.go b/internal/repository/pg_worker_ssh.go index 727f402b..9c1425eb 100644 --- a/internal/repository/pg_worker_ssh.go +++ b/internal/repository/pg_worker_ssh.go @@ -255,7 +255,11 @@ func (r *workerRepository) ListWorkersByProfile(ctx context.Context, profileID u func (r *workerRepository) RecordEnrolledIP(ctx context.Context, id uuid.UUID, ip string) error { _, err := r.db.Exec(ctx, ` - UPDATE workers SET ip_addr = $2, ssh_host = COALESCE(NULLIF(ssh_host,''), $2), updated_at = NOW() + UPDATE workers + SET ip_addr = $2, + ssh_host = COALESCE(NULLIF(ssh_host,''), $2), + install_state = 'provisioning', + updated_at = NOW() WHERE id = $1 `, id, ip) return err diff --git a/scripts/install-worker.sh b/scripts/install-worker.sh index ee71e420..83d6712c 100755 --- a/scripts/install-worker.sh +++ b/scripts/install-worker.sh @@ -24,6 +24,11 @@ # curl -fsSL https://get.warmbly.com/worker | sudo bash -s -- \ # --env-file /root/worker.env # +# Or with a one-time enrollment token from the dashboard: +# +# curl -fsSL https://get.warmbly.com/worker | sudo bash -s -- \ +# --enroll wmenroll_... +# # Re-running is safe: existing env values are preserved unless overridden, # and the worker ID will resolve to the same value as long as the IP is stable. @@ -43,6 +48,8 @@ CONTAINER_NAME="warmbly-worker" ACTION="install" INTERACTIVE=1 SUPPLIED_ENV_FILE="" +ENROLL_TOKEN="" +API_BASE="${WARMBLY_API_BASE:-https://api.warmbly.com}" # Comma-separated list of IPv4 addresses for multi-IP install. When non-empty, # the installer drops one templated systemd unit per IP, each bound to that IP @@ -111,6 +118,9 @@ Configuration flags: --tier Worker tier label (default: shared) --image Docker image (default: ${IMAGE}) --env-file Use this env file verbatim, skip prompts + --enroll Exchange a one-time dashboard enrollment token + for worker config and install without prompts + --api-base API base for --enroll (default: ${API_BASE}) --kafka Kafka bootstrap servers (host:port[,host:port]) --kafka-user @@ -151,6 +161,8 @@ while [[ $# -gt 0 ]]; do --tier) CFG[WORKER_TIER]="$2"; shift 2 ;; --image) IMAGE="$2"; shift 2 ;; --env-file) SUPPLIED_ENV_FILE="$2"; shift 2 ;; + --enroll) ENROLL_TOKEN="$2"; INTERACTIVE=0; shift 2 ;; + --api-base) API_BASE="$2"; shift 2 ;; --kafka) CFG[KAFKA_BOOTSTRAP_SERVERS]="$2"; shift 2 ;; --kafka-user) CFG[KAFKA_SASL_USERNAME]="$2"; shift 2 ;; @@ -315,6 +327,54 @@ merge_existing_env() { done < "$ENV_FILE" } +env_file_value() { + local file="$1" key="$2" + grep -E "^${key}=" "$file" 2>/dev/null | head -1 | cut -d= -f2- +} + +fetch_enrollment_env() { + [[ -n "$ENROLL_TOKEN" ]] || return 1 + command -v curl >/dev/null 2>&1 || die "curl is required for --enroll" + + local ip="${IP_OVERRIDE}" + if [[ -z "$ip" ]]; then + ip="$(detect_public_ip || true)" + fi + + local tmp; tmp="$(mktemp)" + local body + body="{\"token\":\"${ENROLL_TOKEN}\"" + if [[ -n "$ip" ]]; then + body+=",\"public_ip\":\"${ip}\"" + fi + body+="}" + + log "exchanging enrollment token at ${API_BASE}" + curl -fsS \ + -X POST \ + -H "Content-Type: application/json" \ + -H "Accept: text/plain" \ + --data "$body" \ + "${API_BASE%/}/api/v1/workers/enroll" > "$tmp" || { + rm -f "$tmp" + die "enrollment failed" + } + + local worker_id image + worker_id="$(env_file_value "$tmp" WORKER_ID)" + image="$(env_file_value "$tmp" WARMBLY_WORKER_IMAGE)" + [[ -n "$worker_id" ]] || die "enrollment response did not include WORKER_ID" + + install -d -m 0700 "$CONFIG_DIR" + install -m 0600 "$tmp" "$ENV_FILE" + rm -f "$tmp" + + WORKER_ID_OVERRIDE="$worker_id" + [[ -n "$image" ]] && IMAGE="$image" + ok "enrollment config installed" + return 0 +} + write_env_file() { install -d -m 0700 "$CONFIG_DIR" local tmp; tmp="$(mktemp)" @@ -322,10 +382,11 @@ write_env_file() { echo "# Warmbly worker config — managed by install-worker.sh" echo "# $(date -u +%Y-%m-%dT%H:%M:%SZ)" for key in APP_ENV AWS_CONFIG_ENABLED AWS_REGION AWS_ACCESS_KEY_ID AWS_SECRET_ACCESS_KEY \ + ENCRYPTED_KEYS_PROVIDER ENCRYPTED_KEYS_BACKEND_URL ENCRYPTED_KEYS_WORKER_TOKEN \ KAFKA_BOOTSTRAP_SERVERS KAFKA_SASL_USERNAME KAFKA_SASL_PASSWORD \ SCHEMA_REGISTRY_URL SCHEMA_REGISTRY_KEY SCHEMA_REGISTRY_SECRET \ - REDIS WORKER_TIER; do - printf '%s=%s\n' "$key" "${CFG[$key]}" + REDIS EVENTBUS_PROVIDER NATS_URL CODEC_PROVIDER WORKER_TIER WORKER_PUBLIC_IP WORKER_EGRESS_KIND; do + printf '%s=%s\n' "$key" "${CFG[$key]:-}" done for kv in "${EXTRA_ENVS[@]}"; do printf '%s\n' "$kv" @@ -499,6 +560,10 @@ list_installed_instances() { # ---------- actions ---------- prepare_common_env() { + if fetch_enrollment_env; then + return + fi + if [[ -n "$SUPPLIED_ENV_FILE" ]]; then [[ -f "$SUPPLIED_ENV_FILE" ]] || die "env file not found: $SUPPLIED_ENV_FILE" install -d -m 0700 "$CONFIG_DIR" From e35a91f6bc233c37312535c98c85718735a2959f Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 30 May 2026 05:13:18 +0000 Subject: [PATCH 09/11] feat: make worker enrollment default --- deploy/README.md | 11 +++- web/src/app/app/admin/workers/new/page.tsx | 68 +++++++++++++++------- 2 files changed, 54 insertions(+), 25 deletions(-) diff --git a/deploy/README.md b/deploy/README.md index 8758cad3..d5e9adbc 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -61,14 +61,19 @@ Add a worker from the admin dashboard: 1. Provision a VPS, note its public IP + root user 2. Admin → Workers → Add Worker -3. Paste the generated SSH public key into the VPS's `~/.ssh/authorized_keys` -4. Click Test, then Install +3. Copy the generated enrollment command +4. Run it on the VPS as root -The backend SSHes in, uploads `scripts/install-worker.sh`, configures systemd, and starts the worker container. From then on, all lifecycle operations (restart, update, system updates, reboot, rotate keys, logs, uninstall) happen from the dashboard. +The installer exchanges the one-time token for worker config, writes `/etc/warmbly/worker.env`, configures systemd, and starts the worker container. The worker then heartbeats back to the backend and marks itself installed. + +The older SSH-managed path is still supported: paste the generated SSH public key into the VPS's `~/.ssh/authorized_keys`, then click Test and Install. From then on, lifecycle operations (restart, update, system updates, reboot, rotate keys, logs, uninstall) can happen from the dashboard. Manual install on the VPS is also supported: ```bash +sudo bash scripts/install-worker.sh --enroll wmenroll_... + +# or fully manual: sudo bash scripts/install-worker.sh \ --kafka kafka.example.com:9092 \ --schema-registry https://schema.example.com \ diff --git a/web/src/app/app/admin/workers/new/page.tsx b/web/src/app/app/admin/workers/new/page.tsx index 75142201..dd1c7209 100644 --- a/web/src/app/app/admin/workers/new/page.tsx +++ b/web/src/app/app/admin/workers/new/page.tsx @@ -57,7 +57,7 @@ const initialState: WizardState = { risk_pool: "clean", dedicated_user_id: "", dedicated_subscription_id: "", - auto_install: true, + auto_install: false, }; const purposeDefaults: Record> = { @@ -123,6 +123,7 @@ export default function AdminAddWorkerWizard() { ssh_host: state.ssh_host, ssh_port: state.ssh_port, ssh_user: state.ssh_user, + generate_enrollment_token: true, }); setResult(created); append("✓ worker row created"); @@ -134,7 +135,13 @@ export default function AdminAddWorkerWizard() { append("✓ profile assigned"); } - // Stop here unless admin wants to install now too. + if (created.enrollment_token) { + append("ready — run the enrollment command on the VPS"); + setRunning(false); + return; + } + + // Stop here unless admin wants the older SSH install path too. if (!state.auto_install) { append("ready — paste the SSH key into the VPS, then click Test → Install on the detail page"); setRunning(false); @@ -193,7 +200,7 @@ export default function AdminAddWorkerWizard() { const canNext = (() => { switch (step) { case 1: return true; - case 2: return state.ssh_host && state.ssh_port > 0 && preflight?.ok; + case 2: return state.ssh_host && state.ssh_port > 0; case 3: return state.name.length > 0; case 4: return state.purpose !== "dedicated" || (state.dedicated_user_id && state.dedicated_subscription_id); default: return true; @@ -265,10 +272,10 @@ export default function AdminAddWorkerWizard() { {/* Step 2 */} {step === 2 && ( <> - +

- We'll check reachability before creating any database rows, so a typo here - won't leave an orphan worker behind. + This IP is used for the worker record and the one-command enrollment config. + SSH reachability is only needed if you use the older dashboard-driven install path.

@@ -438,21 +445,13 @@ export default function AdminAddWorkerWizard() { )}
- {err &&

{err}

} ) : ( @@ -624,6 +623,9 @@ function PostCreatePanel({ onInstall: () => void; }) { const [pasted, setPasted] = useState(false); + const enrollCommand = result.enrollment_token + ? `curl -fsSL https://get.warmbly.com/worker | sudo bash -s -- --enroll ${result.enrollment_token}` + : ""; return (
@@ -631,6 +633,26 @@ function PostCreatePanel({ ✓ Worker created. ID: {result.id}
+ {result.enrollment_token && ( +
+
Run on the VPS
+
+ {enrollCommand} +
+
+

+ Token expires in {Math.round((result.enrollment_token_ttl_seconds ?? 7200) / 60)} minutes and is consumed after first use. +

+ +
+
+ )} +
SSH public key