Janady13/missionmatch / cmd/oews-national-sync/main.go

1// Command oews-national-sync expands the BLS OEWS wage baseline to ALL detailed SOC occupations at the US 2// NATIONAL level (M5 price to win Phase 1). The existing rates sync (cmd/rates-sync) writes the Galapagos 3// HI and NM state rows; this job writes a national US row for every detailed OEWS occupation so any 4// requirement whose title resolves to a SOC gets at least the national wage band even outside HI and NM. 5// 6// It downloads the BLS oe.occupation flat file (with the required descriptive User Agent that names a 7// contact email, since BLS returns 403 to a default User Agent), keeps the detailed SOCs (display level 8// 3, roughly 800), builds the two national series per SOC (OEUN...04 annual mean, OEUN...13 annual 9// median), pulls the latest annual value in batches of 50 through the shared BLS client, and upserts one 10// row per (occupation_code, area='US', metric) into the global public bls_wages catalog. It is idempotent: 11// a second run updates values and inserts nothing new. Roughly 800 SOCs times 2 datatypes is about 1600 12// series, about 32 batched API requests, well inside the BLS 500 request daily limit. 13// 14// The national rows carry area='US', area_name='United States', and an empty role (national data has no 15// Galapagos role); the HI and NM Galapagos state rows the rates sync writes are complementary and untouched. 16// 17// Env: BLS_API_KEY (server side only, never logged) and MM_MIGRATOR_DATABASE_URL (or MM_DATABASE_URL, the 18// migrator DSN, since bls_wages is a global catalog the app role may only read). Optional: BLS_API_BASE 19// (override the API endpoint), BLS_OCCUPATION_URL (override the flat file URL), BLS_USER_AGENT (override 20// the descriptive User Agent). Build and run FIPS validated: CGO_ENABLED=0 GOFIPS140=v1.0.0 go build, run 21// with GODEBUG=fips140=on. 22package main 23 24import ( 25 "context" 26 "crypto/fips140" 27 "crypto/pbkdf2" 28 "crypto/sha256" 29 "errors" 30 "log" 31 "net/http" 32 "os" 33 "strings" 34 "time" 35 36 "github.com/jackc/pgx/v5/pgxpool" 37 "github.com/shopspring/decimal" 38 39 "github.com/missionworkforce/missionmatch/internal/platform/bls" 40 "github.com/missionworkforce/missionmatch/internal/platform/database" 41 "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" 42) 43 44// defaultUserAgent is the descriptive User Agent naming a contact email, required by the BLS download. 45const defaultUserAgent = "missionMATCH wage sync (joseph.w.anady@gmail.com)" 46 47// batchPause is a small courtesy sleep between BLS API batches so the sync is a good API citizen. 48const batchPause = 500 * time.Millisecond 49 50func main() { 51 ctx, cancel := context.WithTimeout(context.Background(), 20*time.Minute) 52 defer cancel() 53 54 // FIPS self test: derive one PBKDF2 HMAC SHA 256 key under the active crypto module and fail closed if 55 // it errors, so a build that dropped the validated module is caught at boot, the same in build FIPS gate 56 // the api and rates sync binaries use. 57 if err := fipsSelfTest(); err != nil { 58 log.Fatalf("oews-national-sync: FIPS self test (PBKDF2 HMAC SHA 256) failed: %v", err) 59 } 60 log.Printf("oews-national-sync: fips_enabled=%v module=%s", fips140.Enabled(), fips140.Version()) 61 62 key := strings.TrimSpace(os.Getenv("BLS_API_KEY")) 63 dsn := firstNonEmpty(os.Getenv("MM_MIGRATOR_DATABASE_URL"), os.Getenv("MM_DATABASE_URL")) 64 if key == "" { 65 log.Fatal("oews-national-sync: BLS_API_KEY not set (server side env only)") 66 } 67 if dsn == "" { 68 log.Fatal("oews-national-sync: set MM_MIGRATOR_DATABASE_URL (or MM_DATABASE_URL)") 69 } 70 userAgent := firstNonEmpty(os.Getenv("BLS_USER_AGENT"), defaultUserAgent) 71 occURL := firstNonEmpty(os.Getenv("BLS_OCCUPATION_URL"), bls.OccupationListURL) 72 73 pool, err := pgxpool.New(ctx, dsn) 74 if err != nil { 75 log.Fatalf("oews-national-sync: db connect: %v", err) 76 } 77 defer pool.Close() 78 if err := pool.Ping(ctx); err != nil { 79 log.Fatalf("oews-national-sync: db ping: %v", err) 80 } 81 // bls_wages is a global catalog the application role may only read, so this writer needs the BYPASSRLS 82 // migrator DSN; assert it now for a legible failure instead of a mid run permission denied. 83 if err := database.AssertMigratorRole(ctx, pool); err != nil { 84 log.Fatalf("oews-national-sync: %v", err) 85 } 86 q := dbgen.New(pool) 87 88 // Phase 1: download and parse the detailed OEWS occupations (display level 3). 89 occs, err := bls.FetchDetailedOccupations(ctx, &http.Client{Timeout: 90 * time.Second}, occURL, userAgent) 90 if err != nil { 91 log.Fatalf("oews-national-sync: download occupation list: %v", err) 92 } 93 if len(occs) == 0 { 94 log.Fatal("oews-national-sync: the occupation list held no detailed SOCs; refusing to proceed") 95 } 96 log.Printf("oews-national-sync: detailed OEWS occupations found=%d", len(occs)) 97 98 // Phase 2: build the two national series per SOC and index them by series id so the fetched value maps 99 // back to its occupation and metric. 100 byID := make(map[string]bls.SeriesSpec, len(occs)*len(bls.Metrics)) 101 ids := make([]string, 0, len(occs)*len(bls.Metrics)) 102 for _, o := range occs { 103 for _, spec := range bls.NationalSeries(o.SOC, o.Title) { 104 byID[spec.SeriesID] = spec 105 ids = append(ids, spec.SeriesID) 106 } 107 } 108 log.Printf("oews-national-sync: national series to query=%d (about %d batched requests)", len(ids), (len(ids)+49)/50) 109 110 // Phase 3: fetch the latest annual value per series, batched 50 per request with a courtesy pause, then 111 // upsert. FetchLatest already batches at 50; wrap it per batch so a courtesy pause sits between batches 112 // and a No Data series is simply absent (the client skips an empty data series). 113 series, err := fetchBatched(ctx, bls.New(key, os.Getenv("BLS_API_BASE")), ids) 114 if err != nil { 115 if errors.Is(err, bls.ErrThrottled) { 116 log.Printf("oews-national-sync: BLS daily threshold reached, stopping run gracefully") 117 return 118 } 119 log.Fatalf("oews-national-sync: fetch: %v", err) 120 } 121 log.Printf("oews-national-sync: series returned with data=%d", len(series)) 122 123 // Phase 4: upsert one national row per (occupation_code, area='US', metric). A series with No Data is 124 // absent from the results, so it is skipped without failing the run. 125 upserted := 0 126 skippedNoValue := 0 127 for _, s := range series { 128 spec, ok := byID[s.ID] 129 if !ok { 130 continue 131 } 132 var val *decimal.Decimal 133 if v, err := decimal.NewFromString(strings.TrimSpace(s.Value)); err == nil { 134 val = &v 135 } else { 136 skippedNoValue++ 137 continue // an unparseable value is not written; the honest floor holds for that occupation 138 } 139 raw := []byte(`{"series_id":"` + s.ID + `","year":"` + s.Year + `","value":"` + s.Value + `"}`) 140 if err := q.UpsertBlsWageByOcc(ctx, dbgen.UpsertBlsWageByOccParams{ 141 SeriesID: s.ID, 142 OccupationCode: spec.OccCode, 143 OccupationTitle: spec.OccTitle, 144 Role: spec.Role, 145 Area: spec.Area, 146 AreaName: spec.AreaName, 147 Metric: spec.Metric, 148 PeriodYear: s.Year, 149 Value: val, 150 Raw: raw, 151 }); err != nil { 152 log.Printf("oews-national-sync: upsert %s: %v", s.ID, err) 153 continue 154 } 155 upserted++ 156 } 157 158 total := int64(-1) 159 if meta, err := q.WagesSyncMeta(ctx); err == nil { 160 total = meta.Total 161 } 162 log.Printf("oews-national-sync: DONE, occupations=%d series_queried=%d rows_upserted=%d skipped_no_value=%d; store holds %d wage series", 163 len(occs), len(ids), upserted, skippedNoValue, total) 164} 165 166// fetchBatched pulls the latest data point for each series in batches of 50, pausing briefly between 167// batches so the sync is a good API citizen. A No Data series is simply absent from a batch's results. 168func fetchBatched(ctx context.Context, client *bls.Client, ids []string) ([]bls.Series, error) { 169 const batch = 50 170 var out []bls.Series 171 for start := 0; start < len(ids); start += batch { 172 end := start + batch 173 if end > len(ids) { 174 end = len(ids) 175 } 176 got, err := client.FetchLatest(ctx, ids[start:end]) 177 if err != nil { 178 return out, err 179 } 180 out = append(out, got...) 181 if end < len(ids) { 182 select { 183 case <-ctx.Done(): 184 return out, ctx.Err() 185 case <-time.After(batchPause): 186 } 187 } 188 } 189 return out, nil 190} 191 192// fipsSelfTest derives one PBKDF2 HMAC SHA 256 key under the active crypto module. It fails closed if the 193// module is absent or errors, so a non FIPS build is caught at boot. 194func fipsSelfTest() error { 195 _, err := pbkdf2.Key(sha256.New, "oews-national-sync-self-test", []byte("oews-national-sync-salt"), 600000, 32) 196 return err 197} 198 199func firstNonEmpty(vals ...string) string { 200 for _, v := range vals { 201 if s := strings.TrimSpace(v); s != "" { 202 return s 203 } 204 } 205 return "" 206} 207