Janady13/missionmatch / cmd/staffing-proactive-sync/main.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / cmd/staffing-proactive-sync/main.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	cmd/staffing-proactive-sync/main.go
Original SHA-256	c3e79805b895e7206cda04e114fc5cf9215bfaee2bc083617a9bd21b1a14962a
Git blob SHA	3d3045507a8ef671bac5d2ed422089dc3c9e9ff5
Reading-copy SHA-256	c3e79805b895e7206cda04e114fc5cf9215bfaee2bc083617a9bd21b1a14962a

Original source ↗

Recovered source transcript

Download complete source text ↓127 lines · 4.4 KB

// Command staffing-proactive-sync runs the W3 proactive bench watch recompute. For each organization it
// refreshes the proactive staffing recommendations: which available candidate best fits which open
// requirement, superseding that candidate's prior pending proactive rows. It is the scheduled job behind
// the host timer (the same pattern as opp-sync), pure Go and FIPS clean, and safe to rerun: every recompute
// is idempotent and writes a fresh run.
//
// It lists organizations through the migrator role (which sees every org) and does the per org work through
// the NOBYPASSRLS app role under RLS, so one tenant's recompute can never read or write another's. Pass
// -org to limit the run to a single organization.
//
// Env: MM_DATABASE_URL (the app DSN), MM_MIGRATOR_DATABASE_URL (the migrator DSN, for listing orgs; falls
// back to MM_DATABASE_URL). The MEGAMIND plane is not used: proactive rationales are deterministic.
package main

import (
	"context"
	"flag"
	"log"
	"os"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/missionworkforce/missionmatch/internal/modules/matching"
	"github.com/missionworkforce/missionmatch/internal/modules/staffing"
	"github.com/missionworkforce/missionmatch/internal/platform/audit"
	"github.com/missionworkforce/missionmatch/internal/platform/config"
	"github.com/missionworkforce/missionmatch/internal/platform/database"
	"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)

func main() {
	orgFlag := flag.String("org", "", "limit the recompute to a single organization id (default: all orgs)")
	flag.Parse()

	cfg, err := config.Load()
	if err != nil {
		log.Fatalf("staffing-proactive-sync: config: %v", err)
	}
	if cfg.AppDatabaseURL == "" {
		log.Fatal("staffing-proactive-sync: MM_DATABASE_URL (the app DSN) is required")
	}
	migDSN := firstNonEmpty(os.Getenv("MM_MIGRATOR_DATABASE_URL"), cfg.AppDatabaseURL)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
	defer cancel()

	appPool, err := database.NewPool(ctx, cfg.AppDatabaseURL)
	if err != nil {
		log.Fatalf("staffing-proactive-sync: app pool: %v", err)
	}
	defer appPool.Close()
	// Fail closed unless this is the NOBYPASSRLS app role, so the recompute can never bypass tenant RLS.
	if err := database.AssertRole(ctx, appPool, database.AppRoleName()); err != nil {
		log.Fatalf("staffing-proactive-sync: app pool role check: %v", err)
	}
	appDB := tenant.New(appPool)

	// The skill matcher reads the backfilled stored embeddings (the same live path as the API), not request
	// time name embedding. Deterministic rationales for the batch (no inference client), so the cron never
	// hammers the plane.
	svc := staffing.NewService(appDB, matching.NewStoredSkillMatcher(appDB), audit.New(appDB), nil, staffing.Load())

	orgIDs, err := listOrgs(ctx, migDSN, *orgFlag)
	if err != nil {
		log.Fatalf("staffing-proactive-sync: list orgs: %v", err)
	}
	if len(orgIDs) == 0 {
		log.Print("staffing-proactive-sync: no organizations to process")
		return
	}

	total := 0
	for _, orgID := range orgIDs {
		orgCtx := tenant.WithOrg(ctx, orgID.String())
		n, rerr := svc.RecomputeProactive(orgCtx)
		if rerr != nil {
			log.Fatalf("staffing-proactive-sync: org %s: %v", orgID, rerr)
		}
		log.Printf("staffing-proactive-sync: org %s processed %d candidates", orgID, n)
		total += n
	}
	log.Printf("staffing-proactive-sync: done, %d org(s), %d candidate watches refreshed", len(orgIDs), total)
}

// listOrgs returns the organization ids to process. It uses the migrator role, which sees every org, so a
// single recompute run can cover the cluster; the per org work then runs under the app role and RLS.
func listOrgs(ctx context.Context, migDSN, only string) ([]uuid.UUID, error) {
	if only != "" {
		id, err := uuid.Parse(strings.TrimSpace(only))
		if err != nil {
			return nil, err
		}
		return []uuid.UUID{id}, nil
	}
	pool, err := pgxpool.New(ctx, migDSN)
	if err != nil {
		return nil, err
	}
	defer pool.Close()
	rows, err := pool.Query(ctx, "SELECT id FROM organizations ORDER BY id")
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []uuid.UUID
	for rows.Next() {
		var id uuid.UUID
		if err := rows.Scan(&id); err != nil {
			return nil, err
		}
		out = append(out, id)
	}
	return out, rows.Err()
}

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