Janady13/missionmatch / cmd/rates-sync/main.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / cmd/rates-sync/main.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	39df94f92b2a33f852481fe1954dabed841524bb
Branch	w0-aedt-compliance
Path or archive member	cmd/rates-sync/main.go
Original SHA-256	b4503ade8873e4bd4916ba8b76577d3d2d39e53bb41ab88a857deb7935eb09e3
Git blob SHA	795988aa3735d1f2f0459b6bc6e146384ff0fd3f
Reading-copy SHA-256	b4503ade8873e4bd4916ba8b76577d3d2d39e53bb41ab88a857deb7935eb09e3

Original source ↗

Recovered source transcript

Download complete source text ↓112 lines · 3.2 KB

// Command rates-sync pulls the BLS OEWS regional wage baseline for the Galapagos occupations in
// Hawaii and New Mexico into the cached bls_wages table. (CALC ceiling rates were dropped: the public
// CALC API can only return the cheapest 10k of 300k rows with no filter or sort, so it cannot produce
// an accurate band.) One shot job: run with `make sync-rates` or weekly via the systemd timer. Reads
// BLS_API_KEY plus MM_MIGRATOR_DATABASE_URL from env. Logs per area counts and a grand total.
package main

import (
	"context"
	"errors"
	"log"
	"os"
	"strings"

	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/shopspring/decimal"

	"github.com/missionworkforce/missionmatch/internal/platform/bls"
	"github.com/missionworkforce/missionmatch/internal/platform/database/dbgen"
)

func main() {
	ctx := context.Background()
	key := strings.TrimSpace(os.Getenv("BLS_API_KEY"))
	dsn := firstNonEmpty(os.Getenv("MM_MIGRATOR_DATABASE_URL"), os.Getenv("MM_DATABASE_URL"))
	if key == "" {
		log.Fatal("rates-sync: BLS_API_KEY not set (server side env only)")
	}
	if dsn == "" {
		log.Fatal("rates-sync: set MM_MIGRATOR_DATABASE_URL (or MM_DATABASE_URL)")
	}

	pool, err := pgxpool.New(ctx, dsn)
	if err != nil {
		log.Fatalf("rates-sync: db connect: %v", err)
	}
	defer pool.Close()
	if err := pool.Ping(ctx); err != nil {
		log.Fatalf("rates-sync: db ping: %v", err)
	}
	q := dbgen.New(pool)

	client := bls.New(key, os.Getenv("BLS_API_BASE"))
	specs := bls.GalapagosSeries()
	byID := make(map[string]bls.SeriesSpec, len(specs))
	ids := make([]string, 0, len(specs))
	for _, s := range specs {
		byID[s.SeriesID] = s
		ids = append(ids, s.SeriesID)
	}

	log.Printf("rates-sync: provider=BLS OEWS occupations=%d areas=2 series=%d", len(bls.GalapagosOccupations), len(specs))

	series, err := client.FetchLatest(ctx, ids)
	if err != nil {
		if errors.Is(err, bls.ErrThrottled) {
			log.Printf("rates-sync: BLS daily threshold reached, stopping run gracefully")
			return
		}
		log.Fatalf("rates-sync: fetch: %v", err)
	}

	perArea := map[string]int{}
	upserted := 0
	for _, s := range series {
		spec, ok := byID[s.ID]
		if !ok {
			continue
		}
		var val *decimal.Decimal
		if v, err := decimal.NewFromString(strings.TrimSpace(s.Value)); err == nil {
			val = &v
		}
		raw := []byte(`{"series_id":"` + s.ID + `","year":"` + s.Year + `","value":"` + s.Value + `"}`)
		if err := q.UpsertBlsWage(ctx, dbgen.UpsertBlsWageParams{
			SeriesID:        s.ID,
			OccupationCode:  spec.OccCode,
			OccupationTitle: spec.OccTitle,
			Role:            spec.Role,
			Area:            spec.Area,
			AreaName:        spec.AreaName,
			Metric:          spec.Metric,
			PeriodYear:      s.Year,
			Value:           val,
			Raw:             raw,
		}); err != nil {
			log.Printf("rates-sync: upsert %s: %v", s.ID, err)
			continue
		}
		upserted++
		perArea[spec.Area]++
	}

	for _, a := range bls.GalapagosAreas {
		log.Printf("rates-sync: %s (%s): %d wage series", a.Abbr, a.Name, perArea[a.Abbr])
	}
	total := int64(-1)
	if meta, err := q.WagesSyncMeta(ctx); err == nil {
		total = meta.Total
	}
	log.Printf("rates-sync: DONE, upserted %d this run; store holds %d wage series", upserted, total)
}

func firstNonEmpty(vals ...string) string {
	for _, v := range vals {
		if s := strings.TrimSpace(v); s != "" {
			return s
		}
	}
	return ""
}