ultisuite-backend/internal/contacts/discovery/query.go

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
}