diff --git a/internal/app/emailverify/service.go b/internal/app/emailverify/service.go index eacbadac..3cff9d7c 100644 --- a/internal/app/emailverify/service.go +++ b/internal/app/emailverify/service.go @@ -40,6 +40,8 @@ type ProviderSource interface { // ReportVerificationProviderError flips the connection's health when the // provider rejected the key or ran out of credits. ReportVerificationProviderError(ctx context.Context, connectionID uuid.UUID, err error) + // ClearVerificationProviderError withdraws that error once the provider works. + ClearVerificationProviderError(ctx context.Context, connectionID uuid.UUID) } // Service verifies contact email addresses before they are ever sent to. @@ -251,10 +253,22 @@ func (s *service) providerUsable(ctx context.Context, p *Provider) (int, error) s.creditsMu.Unlock() if err != nil { s.noteProviderError(ctx, p, err) + } else { + s.noteProviderOK(ctx, p) } return n, err } +// noteProviderOK withdraws a recorded provider error once the key answers with +// a balance again. The write is guarded in the repository, so a pass that finds +// the connection already healthy costs nothing. +func (s *service) noteProviderOK(ctx context.Context, p *Provider) { + if p == nil || p.ConnectionID == nil || s.providers == nil { + return + } + s.providers.ClearVerificationProviderError(ctx, *p.ConnectionID) +} + func (s *service) VerifyPending(ctx context.Context, limit int) (int, *errx.Error) { cands, xerr := s.repo.ListVerificationCandidates(ctx, limit) if xerr != nil { diff --git a/internal/app/integration/service.go b/internal/app/integration/service.go index e8fa5560..a935a044 100644 --- a/internal/app/integration/service.go +++ b/internal/app/integration/service.go @@ -167,6 +167,7 @@ type Service interface { // emailverify.ProviderSource: the org's paid verification backend, if any. VerificationProviderFor(ctx context.Context, orgID uuid.UUID) (*emailverifyapp.Provider, error) ReportVerificationProviderError(ctx context.Context, connectionID uuid.UUID, err error) + ClearVerificationProviderError(ctx context.Context, connectionID uuid.UUID) // Repo exposes the underlying repository for the inbound webhook handlers. Repo() repository.IntegrationRepository @@ -1433,6 +1434,12 @@ func (s *service) ReportVerificationProviderError(ctx context.Context, connectio _ = s.repo.SetConnectionStatus(ctx, connectionID, status, health, detail) } +// ClearVerificationProviderError puts the connection card back to healthy once +// the provider answers again, since nothing else ever withdrew the error. +func (s *service) ClearVerificationProviderError(ctx context.Context, connectionID uuid.UUID) { + _ = s.repo.ClearConnectionHealth(ctx, connectionID) +} + // slackChannelFor resolves the channel to post org notifications to. The // OAuth connect flow doesn't capture a default channel, so we look (in order) // at the connection's own config, then reuse whatever channel the org already diff --git a/internal/models/integration.go b/internal/models/integration.go index 0dde072f..5446defb 100644 --- a/internal/models/integration.go +++ b/internal/models/integration.go @@ -40,6 +40,10 @@ const ( IntegrationMillionVerifier IntegrationProvider = "millionverifier" ) +// VerificationProviders are the providers that verify contact addresses. A +// connection to one of these makes a built-in verdict worth re-checking. +var VerificationProviders = []IntegrationProvider{IntegrationMillionVerifier} + // AllIntegrationProviders lists every provider the dashboard exposes. The // order here is the catalog order users see. var AllIntegrationProviders = []IntegrationProvider{ diff --git a/internal/repository/contact_verification_reopen_live_test.go b/internal/repository/contact_verification_reopen_live_test.go new file mode 100644 index 00000000..b5a8dd47 --- /dev/null +++ b/internal/repository/contact_verification_reopen_live_test.go @@ -0,0 +1,86 @@ +package repository + +import ( + "context" + "testing" + + "github.com/google/uuid" +) + +// Connecting a paid verifier has to reopen the addresses the built-in check +// could not resolve. Without that, a workspace that connects one because the +// in-house probe returned `unknown` waits out the 30-day shelf life before a +// single credit is spent, and the connection card keeps saying degraded from +// whatever the provider answered on the first pass. +// +// Run against the dev stack: +// +// WARMBLY_TEST_DB=postgres://warmbly:warmbly@localhost:15432/warmbly_dev?sslmode=disable \ +// go test ./internal/repository/ -run LiveVerificationReopen -v +func TestLiveVerificationReopenOnProviderConnect(t *testing.T) { + handle, pool := liveContactDB(t) + f := newSharedOrgFixture(t, pool) + ctx := context.Background() + repo := &contactRepository{DB: handle} + + exec := func(sql string, args ...any) { + t.Helper() + if _, err := pool.Exec(ctx, sql, args...); err != nil { + t.Fatalf("fixture: %v", err) + } + } + + // Checked yesterday by the built-in probe and left unknown: far inside the + // 30-day window, so nothing but a connected verifier makes it a candidate. + stale := uuid.New() + exec(`INSERT INTO contacts (id, user_id, organization_id, email, first_name, last_name, company, phone, + custom_fields, updated_at, created_at, + verification_status, verification_provider, verification_source, verification_checked_at) + VALUES ($1, $2, $3, $4, 'Ada', 'Ng', '', '', '{}'::jsonb, NOW(), NOW(), + 'unknown', 'builtin', 'probe', NOW() - INTERVAL '1 day')`, + stale, f.owner, f.org, "reopen-"+stale.String()[:8]+"@test.local") + + candidates := func() map[uuid.UUID]bool { + t.Helper() + got, xerr := repo.ListVerificationCandidates(ctx, 500) + if xerr != nil { + t.Fatalf("list candidates: %v", xerr) + } + out := map[uuid.UUID]bool{} + for _, c := range got { + out[c.ID] = true + } + return out + } + + if candidates()[stale] { + t.Fatal("a fresh built-in verdict is a candidate with no verifier connected") + } + + conn := uuid.New() + exec(`INSERT INTO integration_connections (id, organization_id, provider, status, health, created_at, updated_at) + VALUES ($1, $2, 'millionverifier', 'connected', 'healthy', NOW(), NOW())`, conn, f.org) + + if !candidates()[stale] { + t.Fatal("connecting a verifier did not reopen the built-in unknown verdict") + } + + // One-shot: a verdict reached after the connection is not reopened again, + // so a provider that keeps answering unknown cannot loop every pass. + exec(`UPDATE contacts SET verification_checked_at = NOW() WHERE id = $1`, stale) + if candidates()[stale] { + t.Fatal("a verdict reached after the connection was reopened again") + } + + // A disconnected verifier is not a reason to re-check anything. + exec(`UPDATE contacts SET verification_checked_at = NOW() - INTERVAL '1 day' WHERE id = $1`, stale) + exec(`UPDATE integration_connections SET status = 'disconnected' WHERE id = $1`, conn) + if candidates()[stale] { + t.Fatal("a disconnected verifier reopened the verdict") + } + + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM contacts WHERE id = $1`, stale) + _, _ = pool.Exec(ctx, `DELETE FROM integration_connections WHERE id = $1`, conn) + }) +} diff --git a/internal/repository/pg_contact.go b/internal/repository/pg_contact.go index b2ce5163..c185321c 100644 --- a/internal/repository/pg_contact.go +++ b/internal/repository/pg_contact.go @@ -697,11 +697,24 @@ func (r *contactRepository) ListVerificationCandidates(ctx context.Context, limi c.verification_checked_at IS NULL OR (c.verification_status = 'unknown' AND c.verification_checked_at < NOW() - make_interval(days => $2)) OR c.verification_checked_at < NOW() - make_interval(days => $3) + -- A verdict reached before the workspace connected a verifier was + -- reached without it; re-check once rather than after the shelf life. + OR EXISTS ( + SELECT 1 FROM integration_connections ic + WHERE ic.organization_id = c.organization_id + AND ic.provider = ANY($5) + AND ic.status <> 'disconnected' + AND c.verification_checked_at < ic.created_at + ) ) ORDER BY c.verification_checked_at ASC NULLS FIRST, c.created_at ASC LIMIT $1 ` - params := []any{limit, config.VerificationUnknownRecheckDays, config.VerificationRecheckDays, config.VerificationEvidenceFreshDays} + providers := make([]string, 0, len(models.VerificationProviders)) + for _, p := range models.VerificationProviders { + providers = append(providers, string(p)) + } + params := []any{limit, config.VerificationUnknownRecheckDays, config.VerificationRecheckDays, config.VerificationEvidenceFreshDays, providers} rows, err := r.DB.Query(ctx, query, params...) if err != nil { db.CaptureError(err, query, params, "query") diff --git a/internal/repository/pg_integration.go b/internal/repository/pg_integration.go index ce8b4a8a..844b290d 100644 --- a/internal/repository/pg_integration.go +++ b/internal/repository/pg_integration.go @@ -61,6 +61,7 @@ type IntegrationRepository interface { MarkConnectionSynced(ctx context.Context, id uuid.UUID, status models.IntegrationStatus, displayFields json.RawMessage, errMsg string) error UpdateConnectionTokens(ctx context.Context, id uuid.UUID, accessEnc, refreshEnc string, expiresAt *time.Time, scopes []string) error SetConnectionStatus(ctx context.Context, id uuid.UUID, status models.IntegrationStatus, health models.IntegrationHealth, detail string) error + ClearConnectionHealth(ctx context.Context, id uuid.UUID) error // OAuth handshake state CreateOAuthState(ctx context.Context, st *models.IntegrationOAuthState) error @@ -350,6 +351,18 @@ func (r *integrationRepository) UpdateConnectionTokens(ctx context.Context, id u return err } +// ClearConnectionHealth marks a connection healthy again, and only writes when +// it is not already, so a per-pass recovery check does not churn the row. +func (r *integrationRepository) ClearConnectionHealth(ctx context.Context, id uuid.UUID) error { + now := time.Now().UTC() + _, err := r.db.Exec(ctx, ` + UPDATE integration_connections + SET status = $1, health = $2, health_detail = NULL, health_checked_at = $3, updated_at = $3 + WHERE id = $4 AND (status <> $1 OR health <> $2)`, + string(models.IntegrationStatusConnected), string(models.IntegrationHealthHealthy), now, id) + return err +} + func (r *integrationRepository) SetConnectionStatus(ctx context.Context, id uuid.UUID, status models.IntegrationStatus, health models.IntegrationHealth, detail string) error { now := time.Now().UTC() _, err := r.db.Exec(ctx, `