Janady13/missionmatch / cmd/alert-sweep/main.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / cmd/alert-sweep/main.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	cmd/alert-sweep/main.go
Original SHA-256	babce7c2dce1400c264ab95dd5819bb0ae2fba30b77a331450f22844498520df
Git blob SHA	7d99a804db55fc4e1d8d0fb0d49c3b5bf7766bc1
Reading-copy SHA-256	babce7c2dce1400c264ab95dd5819bb0ae2fba30b77a331450f22844498520df
Original source ↗
Recovered source transcript

Download complete source text ↓329 lines · 13.1 KB
// Command alert-sweep runs the proactive alert engine, the watch loop, across every organization. For each
// org it runs the three triggers (a clearance about to expire, a winnable recompete that just dropped and
// fits the org's interest, an open gap the bench now structurally covers) and UPSERTs one typed, persisted,
// UI surfaceable alert per detected event, idempotently (a re sweep refreshes rather than duplicates). It is
// the scheduled job behind the host timer (the same pattern as mm-opp-sync), pure Go and FIPS clean, and safe
// to rerun.
//
// It lists organizations through the migrator role (which sees every org) and does the per org sweep through
// the NOBYPASSRLS app role under RLS, so one tenant's sweep can never read or write another's. One org failing
// does not abort the rest. Output is per type COUNTS only, never PII: no candidate id, no agency value, just
// counts. Pass -org to limit the run to a single organization.
//
// Env (server side only): MM_DATABASE_URL (the app DSN, NOBYPASSRLS, required), MM_MIGRATOR_DATABASE_URL (the
// migrator DSN, for listing orgs; falls back to MM_DATABASE_URL).
//
// Build and run FIPS validated: CGO_ENABLED=0 GOFIPS140=v1.0.0 go build, run with GODEBUG=fips140=on.
package main

import (
	"context"
	"crypto/fips140"
	"crypto/pbkdf2"
	"crypto/sha256"
	"flag"
	"fmt"
	"log"
	"os"
	"strings"
	"time"

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

	"github.com/missionworkforce/missionmatch/internal/modules/alerts"
	"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/directory"
	"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)

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

	// FIPS self test: derive one PBKDF2 key under the active crypto module and fail closed if it errors, so a
	// build that dropped the validated module is caught at boot, the same in build FIPS gate the api and the
	// fairness-snapshot binaries use.
	if err := fipsSelfTest(); err != nil {
		log.Fatalf("alert-sweep: FIPS self test (PBKDF2 HMAC SHA 256) failed: %v", err)
	}
	log.Printf("alert-sweep: fips_enabled=%v module=%s", fips140.Enabled(), fips140.Version())

	cfg, err := config.Load()
	if err != nil {
		log.Fatalf("alert-sweep: config: %v", err)
	}
	if cfg.AppDatabaseURL == "" {
		log.Fatal("alert-sweep: 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()

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

	// EACH ORGANIZATION IS SWEPT AGAINST ITS OWN DATABASE, resolved through the tenant directory.
	//
	// This used to open ONE pool on MM_DATABASE_URL and reuse it for every organization. That is correct only
	// when the deployment keeps every tenant in the database that variable names, and this one does not: on
	// the Bubbles node MM_DATABASE_URL is the CONTROL PLANE, while the directory routes the tenant to another
	// machine entirely. The sweep therefore ran nightly, connected to a database holding the organization row
	// and none of its data, found nothing, and reported success. It had been generating no alerts at all.
	//
	// cmd/migrate-all-tenants already had the rule written down: "Silently connecting to the wrong machine and
	// reporting success is the failure this function exists to make impossible." That warning was one command
	// away from the command that needed it.
	//
	// Every org now logs the HOST it was swept against, so a routing mistake is visible in the operator log
	// rather than hiding behind a zero that reads like a quiet night.
	routes, err := resolveRoutes(ctx, migDSN, cfg.AppDatabaseURL, orgIDs)
	if err != nil {
		log.Fatalf("alert-sweep: resolve tenant routes: %v", err)
	}

	var swept int
	var totClearance, totRecompete, totGap, totCollision, totRedeploy int
	for _, orgID := range orgIDs {
		route := routes[orgID]
		sum, host, serr := sweepOne(ctx, orgID, route)
		if serr != nil {
			// One org failing must not abort the schedule for the rest. Log the org id (not PII) and move on.
			log.Printf("alert-sweep: org %s sweep failed: %v", orgID, serr)
			continue
		}
		swept++
		totClearance += sum.ClearanceExpiring
		totRecompete += sum.RecompeteWinnable
		totGap += sum.GapFillable
		totCollision += sum.Collision
		totRedeploy += sum.RedeploymentDue
		// THE LOG ENUMERATES THE TYPES BY HAND, which is how it silently under reported the first time a type
		// was added: the summary struct and its total both knew about redeployment_due and these two lines did
		// not, so the operator log said total=0 while alerts were being written. A count that disagrees with
		// the table is worse than no count, because it is the number somebody checks to see whether the loop
		// ran. Any type added after this one has to be added here too, and the sweep's own test asserts the
		// total, which is the part that cannot be forgotten.
		log.Printf("alert-sweep: org %s swept, clearance_expiring=%d recompete_winnable=%d gap_fillable=%d collision=%d redeployment_due=%d total=%d",
			orgID, sum.ClearanceExpiring, sum.RecompeteWinnable, sum.GapFillable, sum.Collision,
			sum.RedeploymentDue, sum.Total())
		log.Printf("alert-sweep: org %s was swept against %s", orgID, host)
	}
	log.Printf("alert-sweep: done, %d of %d org(s) swept, clearance_expiring=%d recompete_winnable=%d gap_fillable=%d collision=%d redeployment_due=%d",
		swept, len(orgIDs), totClearance, totRecompete, totGap, totCollision, totRedeploy)
}

// listOrgs returns the organization ids to sweep. It uses the migrator role, which sees every org, so a single
// sweep run can cover the cluster; the per org work then runs under the app role and RLS. A passed -org limits
// the run to one organization.
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 ""
}

// fipsSelfTest derives one PBKDF2 HMAC SHA 256 key at a representative cost under the active crypto module. It
// fails closed if the module is absent or errors, so a non FIPS build is caught at boot.
func fipsSelfTest() error {
	_, err := pbkdf2.Key(sha256.New, "alert-sweep-self-test", []byte("alert-sweep-salt"), 600000, 32)
	return err
}

// route is where one organization's data actually lives.
type route struct {
	dsn  string
	host string
}

// resolveRoutes maps every organization to the database the tenant directory names for it.
//
// The directory is the AUTHORITY. If it carries a row for an organization, that row decides the machine, and
// a row without a host is an error rather than an invitation to guess. If the directory is absent entirely,
// which is the single database development case, every org falls back to the configured app DSN and the log
// says so once, plainly, rather than leaving an operator to infer it.
func resolveRoutes(ctx context.Context, migDSN, appDSN string, orgIDs []uuid.UUID) (map[uuid.UUID]route, error) {
	out := make(map[uuid.UUID]route, len(orgIDs))
	fallback := route{dsn: appDSN, host: hostOf(appDSN)}

	// THE DIRECTORY IS NOT RELIABLY ON EITHER ONE OF THESE, so both are tried.
	//
	// tenant_directory lives on the CONTROL PLANE, which on this deployment is the machine MM_DATABASE_URL
	// names, while MM_MIGRATOR_DATABASE_URL points at the tenant machine. Reading it from the migrator DSN
	// alone found no table, fell through to the fallback, and swept the control plane again, which is the
	// original defect wearing a different hat. Trying the app DSN first and the migrator DSN second covers
	// both layouts without the command needing to know which one it is deployed into.
	tenants, err := listTenants(ctx, appDSN, migDSN)
	if err != nil {
		// No directory anywhere is a legitimate single database deployment. Say it once and carry on.
		log.Printf("alert-sweep: no usable tenant directory (%v), sweeping every org against %s", err, fallback.host)
		for _, id := range orgIDs {
			out[id] = fallback
		}
		return out, nil
	}

	byOrg := make(map[uuid.UUID]directory.TenantConn, len(tenants))
	for _, t := range tenants {
		byOrg[t.OrganizationID] = t
	}
	for _, id := range orgIDs {
		t, ok := byOrg[id]
		if !ok {
			out[id] = fallback
			continue
		}
		dsn, err := tenantAppDSN(appDSN, t)
		if err != nil {
			return nil, err
		}
		out[id] = route{dsn: dsn, host: t.Host}
	}
	return out, nil
}

// tenantAppDSN builds the APP role DSN for one tenant, mirroring cmd/migrate-all-tenants exactly so an
// operator learns one naming rule rather than two that look alike. An explicit MM_TENANT_DSN_<REF> wins;
// otherwise the configured app credential is reused against the tenant's own host, port and database.
func tenantAppDSN(appDSN string, t directory.TenantConn) (string, error) {
	if t.SecretRef != "" {
		if override := os.Getenv("MM_TENANT_DSN_" + envKey(t.SecretRef)); override != "" {
			return override, nil
		}
	}
	if t.Host == "" || t.Port == 0 {
		return "", fmt.Errorf("tenant %s has no db_host or db_port in tenant_directory, refusing to guess (set MM_TENANT_DSN_%s or repair the directory row)", t.DBName, envKey(t.SecretRef))
	}
	cfg, err := pgx.ParseConfig(appDSN)
	if err != nil {
		return "", fmt.Errorf("parse app dsn: %w", err)
	}
	out := fmt.Sprintf("host=%s port=%d user=%s dbname=%s", t.Host, t.Port, cfg.User, t.DBName)
	if cfg.Password != "" {
		out += " password=" + cfg.Password
	}
	if cfg.TLSConfig == nil {
		out += " sslmode=disable"
	}
	return out, nil
}

// sweepOne opens a pool on one tenant's database, proves it is the NOBYPASSRLS app role, and sweeps.
//
// The pool is per organization and closed straight after. That costs a connection setup per tenant on a job
// that runs once a day, which is the correct trade for never holding a pool open against the wrong machine.
func sweepOne(ctx context.Context, orgID uuid.UUID, r route) (alerts.SweepSummary, string, error) {
	pool, err := database.NewPool(ctx, r.dsn)
	if err != nil {
		return alerts.SweepSummary{}, r.host, fmt.Errorf("pool on %s: %w", r.host, err)
	}
	defer pool.Close()
	// Fail closed unless this is the NOBYPASSRLS app role, so the sweep can never bypass tenant RLS.
	if err := database.AssertRole(ctx, pool, database.AppRoleName()); err != nil {
		return alerts.SweepSummary{}, r.host, fmt.Errorf("role check on %s: %w", r.host, err)
	}
	db := tenant.New(pool)
	svc := alerts.NewService(db, audit.New(db))
	sum, err := svc.SweepOrg(tenant.WithOrg(ctx, orgID.String()))
	return sum, r.host, err
}

// hostOf reports the host in a DSN for logging. It never returns a credential.
func hostOf(dsn string) string {
	cfg, err := pgx.ParseConfig(dsn)
	if err != nil || cfg.Host == "" {
		return "the configured app database"
	}
	return cfg.Host
}

// envKey folds a secret reference into an environment variable suffix, mirroring
// internal/platform/secrets and cmd/migrate-all-tenants.
func envKey(ref string) string {
	out := make([]rune, 0, len(ref))
	for _, r := range strings.ToUpper(strings.TrimSpace(ref)) {
		if (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') {
			out = append(out, r)
			continue
		}
		out = append(out, '_')
	}
	return string(out)
}

// listTenants reads tenant_directory from the first DSN that has one.
//
// The candidates are tried in order and the first that answers wins. An empty directory counts as no
// directory, because a catalog with no rows cannot route anything and falling back is more useful than
// routing every organization to nowhere.
func listTenants(ctx context.Context, dsns ...string) ([]directory.TenantConn, error) {
	var lastErr error
	for _, dsn := range dsns {
		if strings.TrimSpace(dsn) == "" {
			continue
		}
		pool, err := pgxpool.New(ctx, dsn)
		if err != nil {
			lastErr = err
			continue
		}
		tenants, err := directory.New(pool).List(ctx, directory.StatusActive)
		pool.Close()
		if err != nil {
			lastErr = err
			continue
		}
		if len(tenants) == 0 {
			lastErr = fmt.Errorf("tenant_directory on %s is empty", hostOf(dsn))
			continue
		}
		log.Printf("alert-sweep: routing %d tenant(s) from the directory on %s", len(tenants), hostOf(dsn))
		return tenants, nil
	}
	if lastErr == nil {
		lastErr = fmt.Errorf("no database to read a tenant directory from")
	}
	return nil, lastErr
}