Janady13/missionmatch / internal/modules/federalsignal/service.go

1// Package federalsignal mines public federal records for the deterministic mission and labor classifier. 2// This file is the classification sync pass: it reads the global opportunity_records, usaspending_awards, 3// and sam_opportunities catalogs, runs the pure missionclass classifier over the record text plus, for 4// SAM notices, the fetched attachment text, and upserts mission tags and incumbent labor roles 5// idempotently. The pass is pure of any scoring path and never reaches the eeo schema: mission tags are 6// reference data, never a match input. It writes ONE append only audit event per run carrying counts 7// only, never any PII or record text. 8package federalsignal 9 10import ( 11 "context" 12 "encoding/json" 13 "fmt" 14 "log" 15 16 "github.com/shopspring/decimal" 17 18 "github.com/missionworkforce/missionmatch/internal/platform/audit" 19 "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" 20 "github.com/missionworkforce/missionmatch/internal/platform/missionclass" 21) 22 23// Record table identifiers, pinned to the CHECK constraint in migration 0026 so a typo cannot silently 24// write a row the database would reject. sam_opportunities is mission tagged through its 25// opportunity_records projection, so the SAM path writes only incumbent roles, never a federal_mission_tag. 26const ( 27 tableOpportunityRecords = "opportunity_records" 28 tableUsaspendingAwards = "usaspending_awards" 29 tableSamOpportunities = "sam_opportunities" 30) 31 32// Repo is the narrow data port the service depends on. The real implementation (NewRepo) is dbgen backed 33// and reads the global catalogs as the migrator role; the test injects an in memory fake so the service 34// is exercised with no DB. Every read is ordered by a unique key in the SQL so iteration is byte stable. 35type Repo interface { 36 // AliasPairs returns the (phrase, code) rows that seed the missionclass AliasIndex from real alias and 37 // mission_area skill rows. An empty result is fine: the keyword classifier still tags records. 38 AliasPairs(ctx context.Context) (map[string]string, error) 39 // ListOpportunityRecordsForClassify returns the opportunity_records projection the classifier reads. 40 ListOpportunityRecordsForClassify(ctx context.Context) ([]dbgen.ListOpportunityRecordsForClassifyRow, error) 41 // ListAwardsForClassify returns the usaspending_awards projection the classifier reads. 42 ListAwardsForClassify(ctx context.Context) ([]dbgen.ListAwardsForClassifyRow, error) 43 // ListSamForClassify returns the sam_opportunities projection the attachment fetcher mines. 44 ListSamForClassify(ctx context.Context) ([]dbgen.ListSamForClassifyRow, error) 45 // UpsertFederalMissionTag writes one global mission tag idempotently. 46 UpsertFederalMissionTag(ctx context.Context, p dbgen.UpsertFederalMissionTagParams) error 47 // UpsertAwardIncumbentRole writes one incumbent labor role idempotently. 48 UpsertAwardIncumbentRole(ctx context.Context, p dbgen.UpsertAwardIncumbentRoleParams) error 49} 50 51// SolicitationFetcher fetches and extracts the plain text of a SAM notice's attachments. The real 52// *federalsignal.Fetcher satisfies it; the test injects a fake that returns fixed text with no network. 53type SolicitationFetcher interface { 54 FetchSolicitationText(ctx context.Context, descURL string, links []string) (string, error) 55} 56 57// Auditor records the one platform event for a run. The concrete audit.Writer satisfies it through 58// System; the test uses a fake. Mirrors the oppintel ingest audit shape (a global, NULL org event). 59type Auditor interface { 60 System(ctx context.Context, e audit.Event) error 61} 62 63// Service runs the classification pass. All dependencies are injected so the whole pass is unit testable 64// with no DB and no network. 65type Service struct { 66 repo Repo 67 fetcher SolicitationFetcher 68 auditor Auditor 69 // maxAttachments bounds how many SAM notices the run will fetch attachments for, so one run stays 70 // within the SAM API daily quota. Notices past the cap are counted as skipped, never silently dropped. 71 maxAttachments int 72} 73 74// NewService builds the pass. maxAttachments caps the SAM attachment fetches per run; a value below zero 75// is treated as zero (fetch nothing), a value of zero disables SAM attachment mining for the run. 76func NewService(repo Repo, fetcher SolicitationFetcher, auditor Auditor, maxAttachments int) *Service { 77 if maxAttachments < 0 { 78 maxAttachments = 0 79 } 80 return &Service{repo: repo, fetcher: fetcher, auditor: auditor, maxAttachments: maxAttachments} 81} 82 83// Result is the counted outcome of one run. Every count is derived from rows actually written, never a 84// fabricated number. It is the audit body, so it carries no PII and no record text. 85type Result struct { 86 RecordsScanned int `json:"records_scanned"` 87 MissionTagsWritten int `json:"mission_tags_written"` 88 IncumbentRoles int `json:"incumbent_roles_written"` 89 SamFetched int `json:"sam_attachments_fetched"` 90 SamSkippedCap int `json:"sam_attachments_skipped_cap"` 91} 92 93// confidenceFull is the fixed confidence the deterministic classifier records. The classifier is exact 94// (keyword, curated alias, curated code), so every tag it writes is full confidence, decimal not float so 95// the value is bit stable for the AEDT asset. 96var confidenceFull = decimal.NewFromInt(1) 97 98// ClassifyAll runs the whole pass: it builds the AliasIndex once from real rows, classifies the global 99// records, and upserts mission tags and incumbent roles idempotently, then writes one audit event with 100// counts only. A second run is a no op churn: every upsert is idempotent so no duplicate rows appear. 101func (s *Service) ClassifyAll(ctx context.Context) (Result, error) { 102 var res Result 103 104 pairs, err := s.repo.AliasPairs(ctx) 105 if err != nil { 106 return res, fmt.Errorf("federalsignal: load alias pairs: %w", err) 107 } 108 idx := missionclass.NewAliasIndex(pairs) 109 110 if err := s.classifyOpportunityRecords(ctx, idx, &res); err != nil { 111 return res, err 112 } 113 if err := s.classifyAwards(ctx, idx, &res); err != nil { 114 return res, err 115 } 116 if err := s.classifySam(ctx, &res); err != nil { 117 return res, err 118 } 119 120 if err := s.audit(ctx, res); err != nil { 121 return res, err 122 } 123 return res, nil 124} 125 126// classifyOpportunityRecords tags each opportunity_records row from its title, psc, and naics. The record 127// id (as text) is the record_id the tag table keys on. 128func (s *Service) classifyOpportunityRecords(ctx context.Context, idx missionclass.AliasIndex, res *Result) error { 129 rows, err := s.repo.ListOpportunityRecordsForClassify(ctx) 130 if err != nil { 131 return fmt.Errorf("federalsignal: list opportunity records: %w", err) 132 } 133 for _, r := range rows { 134 res.RecordsScanned++ 135 for _, h := range missionclass.Classify(r.Title, r.Psc, r.Naics, idx) { 136 if err := s.repo.UpsertFederalMissionTag(ctx, dbgen.UpsertFederalMissionTagParams{ 137 RecordTable: tableOpportunityRecords, 138 RecordID: r.RecordID, 139 MissionCode: h.Code, 140 Mechanism: h.Mechanism, 141 Evidence: h.Evidence, 142 Confidence: confidenceFull, 143 }); err != nil { 144 return fmt.Errorf("federalsignal: upsert opportunity mission tag: %w", err) 145 } 146 res.MissionTagsWritten++ 147 } 148 } 149 return nil 150} 151 152// classifyAwards tags each usaspending_awards row from its description plus naics_description, with the 153// naics_code feeding the crosswalk, and mines the incumbent labor mix from the description. The 154// generated_internal_id is the record_id both tables key on. 155func (s *Service) classifyAwards(ctx context.Context, idx missionclass.AliasIndex, res *Result) error { 156 rows, err := s.repo.ListAwardsForClassify(ctx) 157 if err != nil { 158 return fmt.Errorf("federalsignal: list awards: %w", err) 159 } 160 for _, r := range rows { 161 res.RecordsScanned++ 162 missionText := r.Description + " " + r.NaicsDescription 163 for _, h := range missionclass.Classify(missionText, "", r.NaicsCode, idx) { 164 if err := s.repo.UpsertFederalMissionTag(ctx, dbgen.UpsertFederalMissionTagParams{ 165 RecordTable: tableUsaspendingAwards, 166 RecordID: r.RecordID, 167 MissionCode: h.Code, 168 Mechanism: h.Mechanism, 169 Evidence: h.Evidence, 170 Confidence: confidenceFull, 171 }); err != nil { 172 return fmt.Errorf("federalsignal: upsert award mission tag: %w", err) 173 } 174 res.MissionTagsWritten++ 175 } 176 if err := s.writeIncumbentRoles(ctx, tableUsaspendingAwards, r.RecordID, r.Description, res); err != nil { 177 return err 178 } 179 } 180 return nil 181} 182 183// classifySam mines the incumbent labor mix from each SAM notice's attachment text. It does NOT write a 184// federal_mission_tag for a SAM notice: the 0026 CHECK forbids it, and a SAM notice is mission tagged 185// through its opportunity_records projection. Attachment fetches are bounded by maxAttachments so a run 186// stays within the SAM API quota; notices past the cap are counted as skipped, never silently dropped. 187func (s *Service) classifySam(ctx context.Context, res *Result) error { 188 rows, err := s.repo.ListSamForClassify(ctx) 189 if err != nil { 190 return fmt.Errorf("federalsignal: list sam opportunities: %w", err) 191 } 192 for _, r := range rows { 193 res.RecordsScanned++ 194 if res.SamFetched >= s.maxAttachments { 195 res.SamSkippedCap++ 196 continue 197 } 198 links := decodeResourceLinks(r.ResourceLinks) 199 text, ferr := s.fetcher.FetchSolicitationText(ctx, r.DescriptionUrl, links) 200 if ferr != nil { 201 // A cancelled context (or any fetch error) for one notice is best effort: count nothing for it 202 // and continue. The fetcher itself is best effort per attachment, so an error here is a whole 203 // notice failure (for example a cancelled run), never the URL or the api_key, which the fetcher 204 // keeps out of its error strings. 205 log.Printf("federalsignal: sam notice attachment fetch failed, skipping notice") 206 continue 207 } 208 res.SamFetched++ 209 if err := s.writeIncumbentRoles(ctx, tableSamOpportunities, r.RecordID, text, res); err != nil { 210 return err 211 } 212 } 213 return nil 214} 215 216// writeIncumbentRoles classifies the labor mix from text and upserts one award_incumbent_roles row per 217// labor category, with the deterministic count and the first matched phrase as evidence. Empty text or no 218// match writes nothing. 219func (s *Service) writeIncumbentRoles(ctx context.Context, table, recordID, text string, res *Result) error { 220 for _, lh := range missionclass.ClassifyLaborCounts(text) { 221 if err := s.repo.UpsertAwardIncumbentRole(ctx, dbgen.UpsertAwardIncumbentRoleParams{ 222 RecordTable: table, 223 RecordID: recordID, 224 LaborCategory: lh.LCAT, 225 RoleCount: int32(lh.Count), 226 Evidence: lh.Evidence, 227 }); err != nil { 228 return fmt.Errorf("federalsignal: upsert incumbent role: %w", err) 229 } 230 res.IncumbentRoles++ 231 } 232 return nil 233} 234 235// audit writes the single append only event for the run, counts only, no PII and no record text. It 236// mirrors the oppintel ingest audit shape (a global System event with a NULL org). It is fail closed: a 237// run the trail cannot record returns an error, and because every upsert is idempotent a rerun is safe. 238func (s *Service) audit(ctx context.Context, res Result) error { 239 body, err := json.Marshal(res) 240 if err != nil { 241 return fmt.Errorf("federalsignal: marshal classify audit: %w", err) 242 } 243 return s.auditor.System(ctx, audit.Event{ 244 Action: "aedt.federalsignal.classified", 245 EntityType: "federalsignal_sync", 246 EntityID: "classify_all", 247 NewValues: body, 248 }) 249} 250 251// decodeResourceLinks parses the sam_opportunities.resource_links JSON array of strings into a slice. The 252// column is a JSON array (never nil at write, "[]" for empty). A malformed value yields no links rather 253// than a fatal error, so one bad row never aborts the pass. 254func decodeResourceLinks(raw []byte) []string { 255 if len(raw) == 0 { 256 return nil 257 } 258 var links []string 259 if err := json.Unmarshal(raw, &links); err != nil { 260 return nil 261 } 262 return links 263} 264