package discovery import ( "context" "encoding/json" "fmt" "strings" "time" "github.com/jackc/pgx/v5" ) func (s *Service) ListOtherProfiles(ctx context.Context, externalUserID string) ([]Profile, error) { rows, err := s.db.Query(ctx, ` SELECT `+profileSelectColumns+` FROM contact_discovered_profiles p JOIN users u ON p.user_id = u.id WHERE u.external_id = $1 AND p.status = 'suggested' AND NOT p.is_mailing_list AND NOT p.is_disposable AND NOT p.is_spam_heavy`+noReplyProfilesSQL+suggestableProfilesSQL+` ORDER BY `+profileInteractionOrderBy+` `, externalUserID) if err != nil { return nil, err } defer rows.Close() var profiles []Profile for rows.Next() { p, err := scanProfileRow(rows) if err != nil { return nil, err } profiles = append(profiles, p) } if err := rows.Err(); err != nil { return nil, err } s.attachSignaturesToProfiles(ctx, profiles) return profiles, nil } func (s *Service) ListSuggestions(ctx context.Context, externalUserID string, enrichOnly bool) ([]Suggestion, error) { query := ` SELECT s.id::text, COALESCE(s.profile_id::text, ''), COALESCE(s.target_contact_uid, ''), s.suggestion_type, s.field_path, s.suggested_value, COALESCE(s.suggested_label, ''), s.confidence, s.status FROM contact_enrichment_suggestions s JOIN users u ON s.user_id = u.id WHERE u.external_id = $1 AND s.status = 'pending' ` if enrichOnly { query += ` AND s.suggestion_type = 'enrich_contact'` } else { query += ` AND s.suggestion_type IN ('new_contact', 'enrich_contact')` } query += ` ORDER BY s.confidence DESC, s.created_at DESC` rows, err := s.db.Query(ctx, query, externalUserID) if err != nil { return nil, err } defer rows.Close() var suggestions []Suggestion for rows.Next() { var sug Suggestion if err := rows.Scan( &sug.ID, &sug.ProfileID, &sug.TargetContactUID, &sug.SuggestionType, &sug.FieldPath, &sug.SuggestedValue, &sug.SuggestedLabel, &sug.Confidence, &sug.Status, ); err != nil { return nil, err } if sug.ProfileID != "" { p, err := s.getProfileByID(ctx, externalUserID, sug.ProfileID) if err == nil { sug.Profile = &p } } suggestions = append(suggestions, sug) } if err := rows.Err(); err != nil { return nil, err } if len(suggestions) > 0 { ncUserID, bookID := s.resolveDiscoveryNCContext(ctx, externalUserID) contacts := s.loadNCContacts(ctx, ncUserID, bookID) suggestions = filterRedundantSuggestions(suggestions, contacts) } return suggestions, nil } func (s *Service) AcceptProfile(ctx context.Context, externalUserID, profileID, contactUID string) error { ids, err := s.relatedProfileIDs(ctx, externalUserID, profileID) if err != nil { return err } tag, err := s.db.Exec(ctx, ` UPDATE contact_discovered_profiles SET status = 'accepted', linked_contact_uid = NULLIF($3, ''), accepted_at = NOW(), updated_at = NOW() WHERE user_id = (SELECT id FROM users WHERE external_id = $1) AND id = ANY($2::uuid[]) AND status IN ('suggested', 'ignored', 'blocked') `, externalUserID, ids, contactUID) if err != nil { return err } if tag.RowsAffected() == 0 { return fmt.Errorf("profile not found or already processed") } _, _ = s.db.Exec(ctx, ` UPDATE contact_enrichment_suggestions SET status = 'accepted', accepted_at = NOW() WHERE profile_id = ANY($2::uuid[]) AND user_id = (SELECT id FROM users WHERE external_id = $1) AND status = 'pending' `, externalUserID, ids) return nil } func (s *Service) RejectProfile(ctx context.Context, externalUserID, profileID string) error { var email string err := s.db.QueryRow(ctx, ` UPDATE contact_discovered_profiles SET status = 'rejected', rejected_at = NOW(), updated_at = NOW() WHERE id = $2::uuid AND user_id = (SELECT id FROM users WHERE external_id = $1) RETURNING primary_email `, externalUserID, profileID).Scan(&email) if err != nil { if err == pgx.ErrNoRows { return fmt.Errorf("profile not found") } return err } _, _ = s.db.Exec(ctx, ` INSERT INTO contact_discovery_rejections (user_id, rejection_key, rejection_type) VALUES ((SELECT id FROM users WHERE external_id = $1), $2, 'profile') ON CONFLICT DO NOTHING `, externalUserID, "email:"+strings.ToLower(email)) _, _ = s.db.Exec(ctx, ` UPDATE contact_enrichment_suggestions SET status = 'rejected', rejected_at = NOW() WHERE profile_id = $2::uuid AND user_id = (SELECT id FROM users WHERE external_id = $1) AND status = 'pending' `, externalUserID, profileID) return nil } func (s *Service) AcceptSuggestion(ctx context.Context, externalUserID, suggestionID string) (Suggestion, error) { var sug Suggestion err := s.db.QueryRow(ctx, ` UPDATE contact_enrichment_suggestions SET status = 'accepted', accepted_at = NOW() WHERE id = $2::uuid AND user_id = (SELECT id FROM users WHERE external_id = $1) AND status = 'pending' RETURNING id::text, COALESCE(profile_id::text, ''), COALESCE(target_contact_uid, ''), suggestion_type, field_path, suggested_value, COALESCE(suggested_label, ''), confidence, status `, externalUserID, suggestionID).Scan( &sug.ID, &sug.ProfileID, &sug.TargetContactUID, &sug.SuggestionType, &sug.FieldPath, &sug.SuggestedValue, &sug.SuggestedLabel, &sug.Confidence, &sug.Status, ) if err != nil { return Suggestion{}, err } return sug, nil } func (s *Service) RejectSuggestion(ctx context.Context, externalUserID, suggestionID string) error { var profileID, fieldPath, value string err := s.db.QueryRow(ctx, ` UPDATE contact_enrichment_suggestions SET status = 'rejected', rejected_at = NOW() WHERE id = $2::uuid AND user_id = (SELECT id FROM users WHERE external_id = $1) AND status = 'pending' RETURNING COALESCE(profile_id::text, ''), field_path, suggested_value `, externalUserID, suggestionID).Scan(&profileID, &fieldPath, &value) if err != nil { return err } rejKey := fmt.Sprintf("field:%s:%s:%s", profileID, fieldPath, value) _, _ = s.db.Exec(ctx, ` INSERT INTO contact_discovery_rejections (user_id, rejection_key, rejection_type) VALUES ((SELECT id FROM users WHERE external_id = $1), $2, 'field_suggestion') ON CONFLICT DO NOTHING `, externalUserID, rejKey) return nil } func (s *Service) PendingCounts(ctx context.Context, externalUserID string) (otherCount, suggestionCount, ignoredCount, blockedCount int, err error) { otherCount, err = s.countOtherProfileGroups(ctx, externalUserID, "") if err != nil { return } err = s.db.QueryRow(ctx, ` SELECT (SELECT COUNT(*)::int FROM contact_enrichment_suggestions s JOIN users u ON s.user_id = u.id WHERE u.external_id = $1 AND s.status = 'pending'), (SELECT COUNT(*)::int FROM contact_discovered_profiles p JOIN users u ON p.user_id = u.id WHERE u.external_id = $1 AND p.status = 'ignored'), (SELECT COUNT(*)::int FROM contact_discovered_profiles p JOIN users u ON p.user_id = u.id WHERE u.external_id = $1 AND p.status = 'blocked') `, externalUserID).Scan(&suggestionCount, &ignoredCount, &blockedCount) return } func (s *Service) getProfileByID(ctx context.Context, externalUserID, profileID string) (Profile, error) { row := s.db.QueryRow(ctx, ` SELECT `+profileSelectColumns+` FROM contact_discovered_profiles p JOIN users u ON p.user_id = u.id WHERE u.external_id = $1 AND p.id = $2::uuid `, externalUserID, profileID) return scanProfileRow(row) } func (s *Service) loadProfileSignatures(ctx context.Context, profileID string) ([]SignatureEntry, error) { rows, err := s.db.Query(ctx, ` SELECT id::text, COALESCE(message_id::text, ''), signature_text, message_date, confidence FROM contact_discovered_signatures WHERE profile_id = $1::uuid ORDER BY message_date DESC LIMIT 5 `, profileID) if err != nil { return nil, err } defer rows.Close() var sigs []SignatureEntry for rows.Next() { var se SignatureEntry if err := rows.Scan(&se.ID, &se.MessageID, &se.SignatureText, &se.MessageDate, &se.Confidence); err != nil { return nil, err } sigs = append(sigs, se) } return sigs, rows.Err() } type profileScanner interface { Scan(dest ...any) error } func scanProfileRow(row profileScanner) (Profile, error) { var p Profile var allEmailsJSON []byte var enrichedJSON []byte var accountsJSON []byte var lastMsg *time.Time err := row.Scan( &p.ID, &p.DisplayName, &p.PrimaryEmail, &allEmailsJSON, &p.MessageCount, &p.SentCount, &p.ReceivedCount, &p.SpamCount, &p.ForwardedCount, &p.OutboundCount, &p.InboundFromCCCount, &p.CopresenceCCBCCCount, &p.IsMailingList, &p.IsDisposable, &p.IsSpamHeavy, &p.ClassificationReason, &p.LinkedContactUID, &p.EnrichmentStatus, &enrichedJSON, &p.Status, &lastMsg, &accountsJSON, ) if err != nil { return Profile{}, err } _ = json.Unmarshal(allEmailsJSON, &p.AllEmails) _ = json.Unmarshal(accountsJSON, &p.DetectedInAccounts) if len(enrichedJSON) > 0 && string(enrichedJSON) != "null" { var ed EnrichedContactData if json.Unmarshal(enrichedJSON, &ed) == nil { p.EnrichedData = &ed } } p.LastMessageAt = lastMsg return p, nil }