Janady13/missionmatch / internal/modules/governance/service.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / internal/modules/governance/service.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	internal/modules/governance/service.go
Original SHA-256	4919fbcedaf0f0fdae43300bdc35c4970885ed61d306ff3b9cb25d162ea51724
Git blob SHA	8bc700ff5f03d50384725f119f5615eaa53660a5
Reading-copy SHA-256	4919fbcedaf0f0fdae43300bdc35c4970885ed61d306ff3b9cb25d162ea51724
Original source ↗
Recovered source transcript

Download complete source text ↓276 lines · 9.6 KB
// Package governance is the AEDT governance surface: the model version registry, the org AEDT
// inventory, the candidate notice and disclosure record, the retention sweep, and the audit read. It
// runs on the app pool under RLS and never touches the eeo schema (that is the compliance pool's job).
package governance

import (
	"context"
	"encoding/json"
	"errors"

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

	"github.com/missionworkforce/missionmatch/internal/platform/audit"
	"github.com/missionworkforce/missionmatch/internal/platform/database/dbgen"
	"github.com/missionworkforce/missionmatch/internal/platform/tenant"
	"time"
)

// ErrNoDatabase is returned when no app pool is configured.
var ErrNoDatabase = errors.New("governance: no database configured")

// Service implements the AEDT governance operations.
type Service struct {
	db            *tenant.DB
	aud           *audit.Writer
	retentionDays int
}

// NewService builds the service. retentionDays is the configured retention window for identifiable
// screening artifacts after a decision.
func NewService(db *tenant.DB, aud *audit.Writer, retentionDays int) *Service {
	return &Service{db: db, aud: aud, retentionDays: retentionDays}
}

func (s *Service) ok() error {
	if !s.db.Ready() {
		return ErrNoDatabase
	}
	return nil
}

// ListModelVersions returns the scoring model registry: each version with the weights and band cuts it
// used, so a frozen decision is reproducible against the version that produced it.
func (s *Service) ListModelVersions(ctx context.Context) ([]dbgen.AedtModelVersion, error) {
	if err := s.ok(); err != nil {
		return nil, err
	}
	var out []dbgen.AedtModelVersion
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		var e error
		out, e = dbgen.New(tx).ListModelVersions(ctx)
		return e
	})
	return out, err
}

// ListInventory returns the org's register of automated employment decision tools in use.
func (s *Service) ListInventory(ctx context.Context) ([]dbgen.AedtInventory, error) {
	if err := s.ok(); err != nil {
		return nil, err
	}
	var out []dbgen.AedtInventory
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		var e error
		out, e = dbgen.New(tx).ListAedtInventory(ctx)
		return e
	})
	return out, err
}

// InventoryInput registers or updates one AEDT tool in the org inventory.
type InventoryInput struct {
	ToolKey      string `json:"tool_key"`
	DisplayName  string `json:"display_name"`
	Purpose      string `json:"purpose"`
	ModelVersion string `json:"model_version"`
}

// UpsertInventory records a tool in the org inventory and audits the change.
func (s *Service) UpsertInventory(ctx context.Context, in InventoryInput) (dbgen.AedtInventory, error) {
	if err := s.ok(); err != nil {
		return dbgen.AedtInventory{}, err
	}
	var row dbgen.AedtInventory
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		var mv *string
		if in.ModelVersion != "" {
			mv = &in.ModelVersion
		}
		r, e := dbgen.New(tx).UpsertAedtInventory(ctx, dbgen.UpsertAedtInventoryParams{
			ToolKey: in.ToolKey, DisplayName: in.DisplayName, Purpose: in.Purpose, ModelVersion: mv,
		})
		if e != nil {
			return e
		}
		row = r
		return s.audit(ctx, tx, audit.Event{Action: "aedt.inventory.recorded", EntityType: "aedt_tool", EntityID: in.ToolKey})
	})
	return row, err
}

// NoticeInput records that a candidate was given the AEDT notice and disclosure.
type NoticeInput struct {
	NoticeVersion  string   `json:"notice_version"`
	DataCategories []string `json:"data_categories"`
}

// RecordNotice writes the candidate notice and disclosure record on the tenant tx and audits it. The
// candidate consent itself lives in the unreadable eeo schema; this is the public schema disclosure
// that the notice was provided.
func (s *Service) RecordNotice(ctx context.Context, candidateID uuid.UUID, in NoticeInput) (dbgen.AedtCandidateNotice, error) {
	if err := s.ok(); err != nil {
		return dbgen.AedtCandidateNotice{}, err
	}
	cats := in.DataCategories
	if len(cats) == 0 {
		cats = DefaultDataCategories
	}
	catJSON, _ := json.Marshal(cats)
	var row dbgen.AedtCandidateNotice
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		r, e := dbgen.New(tx).InsertCandidateNotice(ctx, dbgen.InsertCandidateNoticeParams{
			CandidateID:    candidateID,
			NoticeVersion:  in.NoticeVersion,
			DataCategories: catJSON,
			RetentionDays:  int32(s.retentionDays),
			DisclosedBy:    actorPtr(ctx),
		})
		if e != nil {
			return e
		}
		row = r
		return s.audit(ctx, tx, audit.Event{Action: "aedt.candidate.notice_disclosed", EntityType: "candidate", EntityID: candidateID.String()})
	})
	return row, err
}

// RunRetention anonymizes the org's candidates whose most recent decision is older than the retention
// window, wiping PII while keeping the de identified outcomes. It returns the number anonymized and
// audits each. This is the on demand, RLS scoped sweep; a platform wide scheduled sweep needs a
// privileged cross org enumeration and is a separate, flagged enhancement.
func (s *Service) RunRetention(ctx context.Context) (int, error) {
	if err := s.ok(); err != nil {
		return 0, err
	}
	count := 0
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		q := dbgen.New(tx)
		due, e := q.RetentionDueCandidates(ctx, int32(s.retentionDays))
		if e != nil {
			return e
		}
		for _, id := range due {
			rows, e := q.AnonymizeCandidate(ctx, id)
			if e != nil {
				return e
			}
			if rows > 0 {
				count++
				// Purge the qualification child data AND the AI memory too, so the retention sweep leaves the
				// same residue as the on demand erasure (sourcing.Delete). Anonymize in place keeps the
				// candidate row, so no FK cascade fires; these deletes are explicit and must mirror the on
				// demand path exactly. The frozen slate_members outcomes are untouched.
				for _, stmt := range []string{
					`DELETE FROM candidate_skills WHERE candidate_id = $1`,
					`DELETE FROM candidate_certifications WHERE candidate_id = $1`,
					`DELETE FROM candidate_customer_experience WHERE candidate_id = $1`,
					`DELETE FROM ai_memory WHERE candidate_id = $1`,
				} {
					if _, e := tx.Exec(ctx, stmt, id); e != nil {
						return e
					}
				}
				// Remove the candidate's W3 staffing recommendations on the retention sweep too (the FK cascade
				// does not fire on anonymize in place, which keeps the candidate row).
				cid := id
				if e := q.DeleteRecommendationsForCandidate(ctx, &cid); e != nil {
					return e
				}
				if e := s.audit(ctx, tx, audit.Event{Action: "retention.candidate.anonymized", EntityType: "candidate", EntityID: id.String()}); e != nil {
					return e
				}
			}
		}
		return nil
	})
	return count, err
}

// ListAudit returns the org's audit trail (org isolated through the GUC bound query, not a Go filter).
func (s *Service) ListAudit(ctx context.Context, limit, offset int32) ([]dbgen.AuditLog, error) {
	if err := s.ok(); err != nil {
		return nil, err
	}
	if limit <= 0 || limit > 500 {
		limit = 100
	}
	var out []dbgen.AuditLog
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		var e error
		out, e = dbgen.New(tx).ListAuditLogsForOrg(ctx, dbgen.ListAuditLogsForOrgParams{Limit: limit, Offset: offset})
		return e
	})
	return out, err
}

// audit records a governance event on the caller's tenant tx and returns the write error so the
// caller fails closed: a mutation that cannot be audited rolls back, matching the auth, sourcing, and
// matching modules and the append only audit contract.
func (s *Service) audit(ctx context.Context, tx pgx.Tx, e audit.Event) error {
	if s.aud == nil {
		return nil
	}
	e.ActorUserID = actorPtr(ctx)
	return s.aud.OnTx(ctx, tx, e)
}

func actorPtr(ctx context.Context) *uuid.UUID {
	if p, ok := tenant.UserFromContext(ctx); ok {
		if id, err := uuid.Parse(p.UserID); err == nil {
			return &id
		}
	}
	return nil
}

// DefaultDataCategories is the AEDT disclosure list: the categories of data the scoring tools assess.
// It is the single source of truth for the candidate notice, reused by the Mission Check result so the
// in modal disclosure names exactly what governance records. None of these is a demographic attribute.
var DefaultDataCategories = []string{
	"clearance self reported indicator",
	"skills",
	"certifications",
	"customer agency experience",
	"years of experience",
	"work location and availability",
}

// NoticeStatus reports whether a candidate has an automated decision notice on file, and the most recent one.
//
// A candidate with NO notice is not an error and not an empty response: it is the answer OnFile false, which
// is what a screen needs in order to prompt for one. Returning a 404 would make "no notice" indistinguishable
// from "no such candidate" and the screen would have to guess.
func (s *Service) NoticeStatus(ctx context.Context, candidateID uuid.UUID) (NoticeStatus, error) {
	var out NoticeStatus
	if err := s.ok(); err != nil {
		return out, err
	}
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		if e := tx.QueryRow(ctx,
			`SELECT count(*)::int FROM aedt_candidate_notices WHERE candidate_id = $1`,
			candidateID).Scan(&out.Count); e != nil {
			return e
		}
		if out.Count == 0 {
			return nil
		}
		out.OnFile = true
		var at time.Time
		var cats []byte
		if e := tx.QueryRow(ctx,
			`SELECT notice_version, data_categories, disclosed_at
			   FROM aedt_candidate_notices
			  WHERE candidate_id = $1
			  ORDER BY disclosed_at DESC LIMIT 1`,
			candidateID).Scan(&out.NoticeVersion, &cats, &at); e != nil {
			return e
		}
		out.DisclosedAt = at.UTC().Format(time.RFC3339)
		_ = json.Unmarshal(cats, &out.Categories)
		return nil
	})
	return out, err
}