Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/missionmatch / internal/modules/alerts/service.go |
| Date | Not specified in source |
| Repository | Janady13/missionmatch |
| Revision | 720d47fae476b0d38477ae4401697d372b1ea2bc |
| Branch | feature/opportunity-to-outcome |
| Path or archive member | internal/modules/alerts/service.go |
| Original SHA-256 | 9cedeed459b1cc37ef8f2c77f228de2b7bb4df7a5bf352c92173c78c268b18b3 |
| Git blob SHA | dfa178e419cbd50facfe8d3334ce6a7bc30fe0c1 |
| Reading-copy SHA-256 | 9cedeed459b1cc37ef8f2c77f228de2b7bb4df7a5bf352c92173c78c268b18b3 |
Recovered source transcript
Download complete source text ↓564 lines · 19.5 KB
package alerts
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/missionworkforce/missionmatch/internal/platform/audit"
"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)
// ErrNoDatabase is returned when no pool is configured. ErrNotFound when a status update matches nothing in
// the caller's org. ErrInvalidStatus when the requested status is outside the allowed set.
var (
ErrNoDatabase = errors.New("alerts: no database configured")
ErrNotFound = errors.New("alerts: alert not found")
ErrInvalidStatus = errors.New("alerts: status must be one of unread, read, dismissed")
)
// The watch loop windows. clearanceWindowDays flags a candidate whose clearance lapses within ninety days,
// the staffing horizon to start a reinvestigation or a replacement. clearanceUrgentDays escalates an alert to
// urgent inside thirty days. recompetePostedWindowDays limits the market arm to opportunities posted in the
// last thirty days (a fresh market event, not stale catalog), and recompeteForwardMonths is the forward
// recompete window: an opportunity whose period of performance ends inside the next twenty four months is a
// takeaway target worth flagging now.
const (
clearanceWindowDays = 90
clearanceUrgentDays = 30
recompetePostedWindowDays = 30
recompeteForwardMonths = 24
// redeployWindowDays is how far ahead the redeployment arm looks. Forty five days is the staffing horizon
// for a contract role: long enough to run a submission and an interview loop before the person is idle,
// short enough that the list stays about people actually rolling off rather than everybody under contract.
redeployWindowDays = 45
// redeployUrgentDays escalates inside two weeks, which is the point at which a replacement assignment
// realistically has to already be in play rather than being started.
redeployUrgentDays = 14
)
// Service runs the proactive watch loop and serves the UI read and status update, all under RLS. The audit
// writer records ONE counts only sweep event with NO PII on the same tenant transaction as the upserts.
type Service struct {
db *tenant.DB
aud *audit.Writer
clock func() time.Time
}
// NewService builds the service. db may be nil (then calls return ErrNoDatabase). The audit writer records the
// counts only sweep event; a nil writer skips the audit (the sweep still runs).
func NewService(db *tenant.DB, aud *audit.Writer) *Service {
return &Service{db: db, aud: aud, clock: func() time.Time { return time.Now().UTC() }}
}
// WithClock overrides the clock, for deterministic tests.
func (s *Service) WithClock(clock func() time.Time) *Service {
s.clock = clock
return s
}
// candidateExpiry, marketOpp, openGap, and collision are the rows the four trigger reads return, the minimal
// owner own data each alert needs. None carries another tenant's data: RLS scopes every read to the caller's
// org plus the global public opportunity_records catalog.
type candidateExpiry struct {
candidateID string
fullName string
daysRemaining int
}
type marketOpp struct {
opportunityRecordID string
agency string
naics string
}
type openGap struct {
requirementID string
title string
}
// collision is one cross opportunity collision row: a candidate proposed or approved on two or more distinct
// competing open opportunities. scarceCleared is a self reported TS SCI active indicator (never a verified
// check), used only to set severity. approvedCount is how many of the competing opportunities the candidate is
// actually approved on (two or more is a true double commitment). suggestedPrimary is the deterministic best fit
// opportunity a human is advised to keep them on; opportunityNames is the ordered list for the detail.
type collision struct {
candidateID string
fullName string
scarceCleared bool
oppCount int
approvedCount int
suggestedPrimary string
opportunityNames string
}
// redeployment is one contractor rolling off: a live assignment ending inside the window with no next
// assignment already lined up. clearanceLevel is carried because it is the first thing a recruiter needs in
// order to know where else this person can go, and it is a self reported indicator like everywhere else.
type redeployment struct {
placementID string
candidateID string
fullName string
roleTitle string
clientName string
daysRemaining int
clearanceLevel string
}
// pendingAlert is one alert the sweep will upsert, assembled in Go from a trigger row before the write.
type pendingAlert struct {
alertType string
severity string
subjectType string
subjectID string // the caller's own record id, empty for none (binds NULL)
title string
detail string
dedupeKey string
}
// SweepOrg runs the four triggers under the caller's tenant context (RLS), UPSERTs one alert per detected
// event ON CONFLICT (organization_id, dedupe_key) so a second sweep changes nothing new (idempotent), and
// writes ONE counts only audit event (aedt.alert.swept) carrying the per type counts and NO PII (no candidate
// id, no agency value, just counts). It returns the per type counts. The whole sweep runs in one tenant
// transaction so the upserts and the audit row commit or roll back together.
func (s *Service) SweepOrg(ctx context.Context) (SweepSummary, error) {
if !s.db.Ready() {
return SweepSummary{}, ErrNoDatabase
}
now := s.clock()
var sum SweepSummary
err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
pending, e := s.collect(ctx, tx, now)
if e != nil {
return e
}
for _, p := range pending {
if e := s.upsert(ctx, tx, p); e != nil {
return e
}
switch p.alertType {
case TypeClearanceExpiring:
sum.ClearanceExpiring++
case TypeRecompeteWinnable:
sum.RecompeteWinnable++
case TypeGapFillable:
sum.GapFillable++
case TypeCollision:
sum.Collision++
case TypeRedeploymentDue:
sum.RedeploymentDue++
}
}
return s.auditSweep(ctx, tx, sum)
})
if err != nil {
return SweepSummary{}, err
}
return sum, nil
}
// collect runs the four trigger reads and assembles the pending alerts in Go, so the write loop is a plain
// UPSERT per alert. Each trigger is RLS scoped (the candidates, requirements, and the org interest read), and
// the market read joins the global public opportunity_records to the org's own interest.
func (s *Service) collect(ctx context.Context, tx pgx.Tx, now time.Time) ([]pendingAlert, error) {
var out []pendingAlert
// today is the date floor for the window math, the DB clock's day. The Go clock is used only to derive
// the cutoff dates the SQL binds; the SQL itself compares against CURRENT_DATE for the day difference, so
// the alert is consistent with the database day.
today := now.Truncate(24 * time.Hour)
clearanceCutoff := today.AddDate(0, 0, clearanceWindowDays)
postedCutoff := today.AddDate(0, 0, -recompetePostedWindowDays)
popEndCutoff := today.AddDate(0, recompeteForwardMonths, 0)
expiries, err := s.readClearanceExpiring(ctx, tx, clearanceCutoff)
if err != nil {
return nil, err
}
for _, e := range expiries {
out = append(out, e.toAlert())
}
opps, err := s.readRecompeteWinnable(ctx, tx, postedCutoff, popEndCutoff)
if err != nil {
return nil, err
}
for _, o := range opps {
out = append(out, o.toAlert())
}
gaps, err := s.readGapFillable(ctx, tx)
if err != nil {
return nil, err
}
for _, g := range gaps {
out = append(out, g.toAlert())
}
collisions, err := s.readCollision(ctx, tx)
if err != nil {
return nil, err
}
for _, cl := range collisions {
out = append(out, cl.toAlert())
}
rolls, err := s.readRedeploymentDue(ctx, tx, today.AddDate(0, 0, redeployWindowDays))
if err != nil {
return nil, err
}
for _, rd := range rolls {
out = append(out, rd.toAlert())
}
return out, nil
}
// readRedeploymentDue runs trigger (e).
func (s *Service) readRedeploymentDue(ctx context.Context, tx pgx.Tx, cutoff time.Time) ([]redeployment, error) {
rows, err := tx.Query(ctx, redeploymentDueSQL, cutoff)
if err != nil {
return nil, fmt.Errorf("alerts: read redeployment due: %w", err)
}
defer rows.Close()
out := make([]redeployment, 0, 32)
for rows.Next() {
var r redeployment
if e := rows.Scan(&r.placementID, &r.candidateID, &r.fullName, &r.roleTitle, &r.clientName,
&r.daysRemaining, &r.clearanceLevel); e != nil {
return nil, fmt.Errorf("alerts: scan redeployment due: %w", e)
}
out = append(out, r)
}
return out, rows.Err()
}
// toAlert turns a rolling off assignment into a pending alert.
//
// THE SUBJECT IS THE CANDIDATE, not the placement, and that is deliberate. What a recruiter does with this
// alert is go and find the person another assignment, so the alert has to open onto the person. It also means
// the dedupe key survives a placement being corrected and reissued, which would otherwise produce a second
// alert about the same human.
func (r redeployment) toAlert() pendingAlert {
severity := SeverityWatch
if r.daysRemaining <= redeployUrgentDays {
severity = SeverityUrgent
}
where := r.roleTitle
if r.clientName != "" {
if where != "" {
where += " at " + r.clientName
} else {
where = r.clientName
}
}
detail := fmt.Sprintf("%d days left", r.daysRemaining)
if where != "" {
detail = where + ", " + detail
}
// The clearance is named because it is the first thing that decides where else this person can go, and a
// recruiter reading a list of names without it has to open every one.
if r.clearanceLevel != "" && r.clearanceLevel != "none" && r.clearanceLevel != "unknown" {
detail += ", " + r.clearanceLevel
}
return pendingAlert{
alertType: TypeRedeploymentDue,
severity: severity,
subjectType: SubjectCandidate,
subjectID: r.candidateID,
title: fmt.Sprintf("%s rolls off in %d days", r.fullName, r.daysRemaining),
detail: detail,
dedupeKey: dedupeKey(TypeRedeploymentDue, r.candidateID),
}
}
// readClearanceExpiring runs trigger (a) and scans the rows. clearanceCutoff is today plus the window in days.
func (s *Service) readClearanceExpiring(ctx context.Context, tx pgx.Tx, clearanceCutoff time.Time) ([]candidateExpiry, error) {
rows, err := tx.Query(ctx, clearanceExpiringSQL, clearanceCutoff)
if err != nil {
return nil, err
}
defer rows.Close()
var out []candidateExpiry
for rows.Next() {
var c candidateExpiry
if err := rows.Scan(&c.candidateID, &c.fullName, &c.daysRemaining); err != nil {
return nil, err
}
out = append(out, c)
}
return out, rows.Err()
}
// readRecompeteWinnable runs trigger (b), the market arm. postedCutoff is today minus the recent window;
// popEndCutoff is today plus the forward recompete window.
func (s *Service) readRecompeteWinnable(ctx context.Context, tx pgx.Tx, postedCutoff, popEndCutoff time.Time) ([]marketOpp, error) {
rows, err := tx.Query(ctx, recompeteWinnableSQL, postedCutoff, popEndCutoff)
if err != nil {
return nil, err
}
defer rows.Close()
var out []marketOpp
for rows.Next() {
var o marketOpp
if err := rows.Scan(&o.opportunityRecordID, &o.agency, &o.naics); err != nil {
return nil, err
}
out = append(out, o)
}
return out, rows.Err()
}
// readGapFillable runs trigger (c), the structural coverage test.
func (s *Service) readGapFillable(ctx context.Context, tx pgx.Tx) ([]openGap, error) {
rows, err := tx.Query(ctx, gapFillableSQL)
if err != nil {
return nil, err
}
defer rows.Close()
var out []openGap
for rows.Next() {
var g openGap
if err := rows.Scan(&g.requirementID, &g.title); err != nil {
return nil, err
}
out = append(out, g)
}
return out, rows.Err()
}
// readCollision runs trigger (d), the cross opportunity collision read, and scans the rows.
func (s *Service) readCollision(ctx context.Context, tx pgx.Tx) ([]collision, error) {
rows, err := tx.Query(ctx, collisionSQL)
if err != nil {
return nil, err
}
defer rows.Close()
var out []collision
for rows.Next() {
var c collision
if err := rows.Scan(&c.candidateID, &c.fullName, &c.scarceCleared, &c.oppCount, &c.approvedCount, &c.suggestedPrimary, &c.opportunityNames); err != nil {
return nil, err
}
out = append(out, c)
}
return out, rows.Err()
}
// toAlert turns a collision row into a pending alert. Severity is urgent when the candidate is a scarce cleared
// person (a self reported TS SCI active indicator) OR is actually approved on two or more competing bids (a
// true double commitment); otherwise watch. The subject is the owner's own candidate; the detail names the
// competing opportunities and the deterministic best fit suggested primary, and states plainly that clearance
// is self reported intelligence, not verified, and that the human resolves the assignment (never auto rejected).
func (c collision) toAlert() pendingAlert {
severity := SeverityWatch
if c.scarceCleared || c.approvedCount >= 2 {
severity = SeverityUrgent
}
who := c.fullName
if c.scarceCleared {
who = "Scarce cleared " + c.fullName
}
title := fmt.Sprintf("%s is on %d competing open bids", who, c.oppCount)
detail := fmt.Sprintf("Proposed or approved across: %s. Suggested primary (best fit): %s.", c.opportunityNames, c.suggestedPrimary)
if c.approvedCount >= 2 {
detail += fmt.Sprintf(" Approved on %d of them, a double commitment.", c.approvedCount)
}
detail += " Advisory: clearance is self reported intelligence, not verified. You resolve the assignment."
return pendingAlert{
alertType: TypeCollision,
severity: severity,
subjectType: SubjectCandidate,
subjectID: c.candidateID,
title: title,
detail: detail,
dedupeKey: dedupeKey(TypeCollision, c.candidateID),
}
}
// toAlert turns a clearance expiry row into a pending alert. The severity is urgent inside the urgent window,
// otherwise watch. The title names the owner's own candidate and the days remaining, the owner's own data.
func (c candidateExpiry) toAlert() pendingAlert {
severity := SeverityWatch
if c.daysRemaining <= clearanceUrgentDays {
severity = SeverityUrgent
}
return pendingAlert{
alertType: TypeClearanceExpiring,
severity: severity,
subjectType: SubjectCandidate,
subjectID: c.candidateID,
title: fmt.Sprintf("%s clearance expires in %d days", c.fullName, c.daysRemaining),
detail: fmt.Sprintf("%d days remaining", c.daysRemaining),
dedupeKey: dedupeKey(TypeClearanceExpiring, c.candidateID),
}
}
// toAlert turns a market opportunity row into a pending alert. It is a market event, watch severity, naming
// the agency and naics the owner already tracks.
func (o marketOpp) toAlert() pendingAlert {
title := fmt.Sprintf("Winnable recompete at %s", o.agency)
detail := ""
if o.naics != "" {
detail = "NAICS " + o.naics
}
return pendingAlert{
alertType: TypeRecompeteWinnable,
severity: SeverityWatch,
subjectType: SubjectOpportunity,
subjectID: o.opportunityRecordID,
title: title,
detail: detail,
dedupeKey: dedupeKey(TypeRecompeteWinnable, o.opportunityRecordID),
}
}
// toAlert turns an open structurally fillable gap into a pending alert. The bench now covers it, so it is an
// actionable opening at watch severity, naming the owner's own requirement.
func (g openGap) toAlert() pendingAlert {
return pendingAlert{
alertType: TypeGapFillable,
severity: SeverityWatch,
subjectType: SubjectRequirement,
subjectID: g.requirementID,
title: fmt.Sprintf("Bench now covers %s", g.title),
detail: "the bench structurally covers this open requirement",
dedupeKey: dedupeKey(TypeGapFillable, g.requirementID),
}
}
// upsert writes one pending alert ON CONFLICT (organization_id, dedupe_key). subjectID empty binds NULL. The
// organization_id is the session GUC inside the SQL, never a caller argument.
func (s *Service) upsert(ctx context.Context, tx pgx.Tx, p pendingAlert) error {
var subjectID *uuid.UUID
if p.subjectID != "" {
id, err := uuid.Parse(p.subjectID)
if err != nil {
return err
}
subjectID = &id
}
_, err := tx.Exec(ctx, upsertAlertSQL, p.alertType, p.severity, p.subjectType, subjectID, p.title, p.detail, p.dedupeKey)
return err
}
// auditSweep writes ONE counts only AEDT event for the sweep, carrying NO PII: the per type counts and the
// total, never a candidate id, an agency value, or a requirement title. The actor is taken from the validated
// session, never a caller argument. A nil writer skips the audit (the sweep still ran).
func (s *Service) auditSweep(ctx context.Context, tx pgx.Tx, sum SweepSummary) error {
if s.aud == nil {
return nil
}
payload, _ := json.Marshal(map[string]any{
"clearance_expiring": sum.ClearanceExpiring,
"recompete_winnable": sum.RecompeteWinnable,
"gap_fillable": sum.GapFillable,
"collision": sum.Collision,
"redeployment_due": sum.RedeploymentDue,
"total": sum.Total(),
})
e := audit.Event{Action: "aedt.alert.swept", EntityType: "alert_sweep", NewValues: payload}
if p, ok := tenant.UserFromContext(ctx); ok {
if uid, perr := uuid.Parse(p.UserID); perr == nil {
e.ActorUserID = &uid
}
}
return s.aud.OnTx(ctx, tx, e)
}
// ListAlerts returns the caller's alerts newest first under RLS, optionally filtered by status. An empty
// status returns every status; a non empty status is validated against the closed set first.
func (s *Service) ListAlerts(ctx context.Context, status string) ([]Alert, error) {
if status != "" && !IsValidStatus(status) {
return nil, ErrInvalidStatus
}
if !s.db.Ready() {
return nil, ErrNoDatabase
}
var out []Alert
err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
rows, e := tx.Query(ctx, listAlertsSQL, status)
if e != nil {
return e
}
defer rows.Close()
for rows.Next() {
a, se := scanAlert(rows)
if se != nil {
return se
}
out = append(out, a)
}
return rows.Err()
})
if err != nil {
return nil, err
}
return out, nil
}
// SetAlertStatus flips one alert's status under RLS (the UI mark read or dismiss). The status is validated
// against the closed set before the write so a bad value is a clean typed error, never a raw constraint
// violation. An id this tenant cannot see returns ErrNotFound, never another org's row.
func (s *Service) SetAlertStatus(ctx context.Context, id uuid.UUID, status string) (Alert, error) {
if !IsValidStatus(status) {
return Alert{}, ErrInvalidStatus
}
if !s.db.Ready() {
return Alert{}, ErrNoDatabase
}
var out Alert
err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
row := tx.QueryRow(ctx, setAlertStatusSQL, id, status)
a, se := scanAlert(row)
if errors.Is(se, pgx.ErrNoRows) {
return ErrNotFound
}
if se != nil {
return se
}
out = a
return nil
})
if err != nil {
return Alert{}, err
}
return out, nil
}
// scanAlert scans one alerts row into the DTO. The subject id is a COALESCE'd string in the SQL, so an empty
// string means no subject and the JSON omits it.
func scanAlert(row pgx.Row) (Alert, error) {
var (
id string
alertType string
severity string
subjectType string
subjectID string
title string
detail string
status string
createdAt time.Time
updatedAt time.Time
)
if err := row.Scan(&id, &alertType, &severity, &subjectType, &subjectID, &title, &detail, &status, &createdAt, &updatedAt); err != nil {
return Alert{}, err
}
return Alert{
ID: id,
AlertType: alertType,
Severity: severity,
SubjectType: subjectType,
SubjectID: subjectID,
Title: title,
Detail: detail,
Status: status,
CreatedAt: createdAt.Format(time.RFC3339),
UpdatedAt: updatedAt.Format(time.RFC3339),
}, nil
}