Janady13/missionmatch / internal/platform/sourcing/ingest.go

1package sourcing 2 3import ( 4 "context" 5 "errors" 6 "fmt" 7 "strings" 8 9 "github.com/google/uuid" 10 "github.com/jackc/pgx/v5" 11 12 "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" 13 "github.com/missionworkforce/missionmatch/internal/platform/extraction" 14) 15 16// IngestPerson resolves a sourced person to a candidate record and writes their STRUCTURED workforce 17// data, so a sourced lead is a real scoring input (skills, certs, customer agencies as rows), not a free 18// text headline, and never a duplicate of an existing bench person. It runs on the caller's transaction; 19// the import command wraps it with the sourcing_log (consent and legal basis) write. Returns the 20// resolved candidate id. 21// 22// Identity resolution uses blocking plus multi attribute matching: an exact email match merges into the 23// existing person; failing that, an exact normalized name plus location match merges; otherwise a new 24// external candidate is created. A merge ENRICHES (links the external id, backfills an empty headline and 25// a lower years figure, adds any new skills, certs, and agencies) and never clobbers curated bench data. 26func IngestPerson(ctx context.Context, q *dbgen.Queries, orgID uuid.UUID, sp SourcedPerson, ext extraction.Extractor) (uuid.UUID, error) { 27 candID, _, err := writeStructuredCandidate(ctx, q, orgID, sp, ext) 28 return candID, err 29} 30 31// writeStructuredCandidate is the shared body of the sourced lead and the bench record ingest paths. It 32// resolves identity (email then name plus location blocking), upserts or enriches the candidate THROUGH 33// the same clearance hardcoding path (UpsertExternalCandidate writes clearance_level 'unknown', 34// clearance_poly 'none', clearance_status 'unknown', so a sourced or bench clearance code can never score 35// as a held clearance), and writes the structured skills, certs, and agencies verbatim. It returns the 36// resolved candidate id and whether the row was an existing person (an enrich) or a new insert. Both the 37// sourced lead path and the bench record path call this, so the dedup and the structured writes are 38// byte identical between them and IngestPerson's signature is unchanged for its existing caller. 39func writeStructuredCandidate(ctx context.Context, q *dbgen.Queries, orgID uuid.UUID, sp SourcedPerson, ext extraction.Extractor) (candID uuid.UUID, found bool, err error) { 40 candID, found, err = resolveCandidate(ctx, q, orgID, sp) 41 if err != nil { 42 return uuid.Nil, false, err 43 } 44 45 if found { 46 if err := q.LinkSourcedCandidate(ctx, dbgen.LinkSourcedCandidateParams{ 47 ID: candID, ExternalID: nzPtr(sp.ExternalID), Headline: nzPtr(buildHeadline(sp)), YearsExperience: int32(sp.Years), 48 }); err != nil { 49 return uuid.Nil, false, fmt.Errorf("link sourced candidate: %w", err) 50 } 51 } else { 52 // ExternalID is NULL when empty (nzPtr), never the empty string. The candidates partial unique index 53 // is (organization_id, external_id) WHERE external_id IS NOT NULL, so an empty string would collapse 54 // every external id free bench record onto one (org, '') row. A bench file usually carries no external 55 // id, so identity for those records is resolved only by the email then name plus location dedup above, 56 // and a genuinely new person inserts a fresh row. 57 candID, err = q.UpsertExternalCandidate(ctx, dbgen.UpsertExternalCandidateParams{ 58 OrganizationID: orgID, 59 FullName: sp.FullName, 60 Email: firstOrNil(sp.Emails), 61 Headline: nzPtr(buildHeadline(sp)), 62 YearsExperience: int32(sp.Years), 63 Location: nzPtr(sp.Location), 64 Remote: strings.Contains(strings.ToLower(sp.Location), "remote"), 65 ExternalID: nzPtr(sp.ExternalID), 66 }) 67 if err != nil { 68 return uuid.Nil, false, fmt.Errorf("upsert external candidate: %w", err) 69 } 70 } 71 72 // Skills come from the provider's STRUCTURED field, more reliable than gazetteer extraction. Each is 73 // upserted into the ontology then attached; the add queries are idempotent (ON CONFLICT DO NOTHING). 74 for _, name := range dedupe(sp.Skills) { 75 sk, err := q.UpsertSkill(ctx, dbgen.UpsertSkillParams{Name: name}) 76 if err != nil { 77 return uuid.Nil, false, fmt.Errorf("upsert skill: %w", err) 78 } 79 if err := q.AddCandidateSkill(ctx, dbgen.AddCandidateSkillParams{OrganizationID: orgID, CandidateID: candID, SkillID: sk.ID}); err != nil { 80 return uuid.Nil, false, fmt.Errorf("add candidate skill: %w", err) 81 } 82 } 83 84 // Certifications and customer agencies are not structured in the provider feed, so extract them from 85 // the profile text via the extraction port. Its skills output is ignored in favor of sp.Skills, and its 86 // ClearanceLevel output is intentionally ignored too, so a resume regex never reaches a clearance column. 87 rx, err := ext.ExtractResume(ctx, profileText(sp)) 88 if err != nil { 89 return uuid.Nil, false, fmt.Errorf("extract profile: %w", err) 90 } 91 for _, name := range dedupe(rx.Certifications) { 92 if err := q.AddCandidateCertification(ctx, dbgen.AddCandidateCertificationParams{OrganizationID: orgID, CandidateID: candID, Name: name}); err != nil { 93 return uuid.Nil, false, fmt.Errorf("add candidate certification: %w", err) 94 } 95 } 96 for _, agency := range dedupe(rx.Agencies) { 97 if err := q.AddCandidateCustomerExperience(ctx, dbgen.AddCandidateCustomerExperienceParams{OrganizationID: orgID, CandidateID: candID, Agency: agency}); err != nil { 98 return uuid.Nil, false, fmt.Errorf("add candidate customer experience: %w", err) 99 } 100 } 101 return candID, found, nil 102} 103 104// resolveCandidate returns an existing candidate id for the sourced person when identity resolution 105// matches, else found is false. Email is tried first (strongest signal), then name plus a non empty 106// location. ErrNoRows is the normal not found path, not an error. 107func resolveCandidate(ctx context.Context, q *dbgen.Queries, orgID uuid.UUID, sp SourcedPerson) (id uuid.UUID, found bool, err error) { 108 if len(sp.Emails) > 0 && strings.TrimSpace(sp.Emails[0]) != "" { 109 got, e := q.FindCandidateByEmail(ctx, dbgen.FindCandidateByEmailParams{OrganizationID: orgID, Lower: sp.Emails[0]}) 110 if e == nil { 111 return got, true, nil 112 } 113 if !errors.Is(e, pgx.ErrNoRows) { 114 return uuid.Nil, false, fmt.Errorf("find by email: %w", e) 115 } 116 } 117 if strings.TrimSpace(sp.FullName) != "" && strings.TrimSpace(sp.Location) != "" { 118 got, e := q.FindCandidateByNameLocation(ctx, dbgen.FindCandidateByNameLocationParams{OrganizationID: orgID, Lower: sp.FullName, Lower_2: sp.Location}) 119 if e == nil { 120 return got, true, nil 121 } 122 if !errors.Is(e, pgx.ErrNoRows) { 123 return uuid.Nil, false, fmt.Errorf("find by name and location: %w", e) 124 } 125 } 126 return uuid.Nil, false, nil 127} 128 129// buildHeadline composes a self reported title and skills line for a sourced lead. 130func buildHeadline(sp SourcedPerson) string { 131 skills := sp.Skills 132 if len(skills) > 6 { 133 skills = skills[:6] 134 } 135 h := sp.JobTitle 136 if h == "" { 137 h = sp.Role 138 } 139 if len(skills) > 0 { 140 if h != "" { 141 h += " | " 142 } 143 h += strings.Join(skills, ", ") 144 } 145 return strings.TrimSpace(h) 146} 147 148// profileText is the free text the extractor reads for certs and agencies (skills come structured). 149func profileText(sp SourcedPerson) string { 150 return strings.Join([]string{buildHeadline(sp), strings.Join(sp.Skills, " "), sp.JobTitle, sp.Role}, " ") 151} 152 153func nzPtr(s string) *string { 154 s = strings.TrimSpace(s) 155 if s == "" { 156 return nil 157 } 158 return &s 159} 160 161func firstOrNil(xs []string) *string { 162 if len(xs) > 0 { 163 return nzPtr(xs[0]) 164 } 165 return nil 166} 167 168func dedupe(xs []string) []string { 169 seen := map[string]bool{} 170 out := make([]string, 0, len(xs)) 171 for _, x := range xs { 172 t := strings.TrimSpace(x) 173 if t == "" { 174 continue 175 } 176 k := strings.ToLower(t) 177 if seen[k] { 178 continue 179 } 180 seen[k] = true 181 out = append(out, t) 182 } 183 return out 184} 185