Janady13/missionmatch / internal/modules/recruiting/analytics_service.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / internal/modules/recruiting/analytics_service.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	internal/modules/recruiting/analytics_service.go
Original SHA-256	9d7aa10ae0ca1e2fc42e0c813c9566cf8be49dffe47366c2520dbd020c3f5cdc
Git blob SHA	c8dbae50d3808d08240efda9b15c8731ebebd6e5
Reading-copy SHA-256	9d7aa10ae0ca1e2fc42e0c813c9566cf8be49dffe47366c2520dbd020c3f5cdc
Original source ↗
Recovered source transcript

Download complete source text ↓314 lines · 12.0 KB
package recruiting

import (
	"context"
	"time"

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

// AnalyticsView is the recruiting report. Every number traces to a row a human created.
type AnalyticsView struct {
	Since    string `json:"since"`
	TimeZone string `json:"time_zone"`

	// TimeToOffer and TimeToStart are the two clocks the industry uses and does not agree on. Reporting
	// one silently is how two tools describe the same desk differently, so both are here and named:
	// opened until an offer was accepted, and opened until someone actually started.
	TimeToOffer TimeToFill `json:"time_to_offer"`
	TimeToStart TimeToFill `json:"time_to_start"`
	// WorkedTimeToStart subtracts time the order spent on hold, using the exact accumulator from
	// migration 0052. A client freezing a program is not the recruiter being slow.
	WorkedTimeToStart TimeToFill `json:"worked_time_to_start"`

	FillRate FillRate      `json:"fill_rate"`
	Aging    []AgingBucket `json:"aging"`
	// OldestOpenDays is the tail the aging buckets summarise. An average would hide it.
	OldestOpenDays int `json:"oldest_open_days"`

	Funnel     []FunnelStage `json:"funnel"`
	Conversion []Conversion  `json:"conversion"`

	// DeskOutcomes is what each recruiter PRODUCED, beside the effort report that says what they DID. A desk
	// logging two hundred calls and placing nobody reads identically to one logging forty and placing four in
	// an effort report, and differently here.
	DeskOutcomes []DeskOutcome `json:"desk_outcomes"`
	// Redeployment is the share of contractors finishing an assignment who were placed again. Nothing could
	// compute it until the redeployment arm existed, and it is the one metric that measures whether a firm
	// keeps the people it already won.
	Redeployment RedeploymentRate `json:"redeployment"`

	Sources    []SourceEffect    `json:"sources"`
	Recruiters []RecruiterEffort `json:"recruiters"`

	Placements PlacementSummary `json:"placements"`
}

// SourceEffect is how a sourcing channel actually performed. Applications and placements are reported
// separately with a rate, because a channel that produces many candidates and no placements is a cost,
// not a success, and a raw candidate count cannot tell the two apart.
type SourceEffect struct {
	Source       string `json:"source"`
	Applications int    `json:"applications"`
	Placed       int    `json:"placed"`
	// Rate is absent below the reporting floor, because one placement from one candidate is a hundred
	// percent and means nothing.
	Rate       string `json:"rate,omitempty"`
	Suppressed bool   `json:"suppressed"`
}

// PlacementSummary is the revenue side. Contract and permanent are never blended: an hourly spread and a
// one time fee are different quantities and a figure averaging them describes neither.
type PlacementSummary struct {
	Live            int `json:"live"`
	Started         int `json:"started"`
	Ended           int `json:"ended"`
	FellOff         int `json:"fell_off"`
	GuaranteeBreach int `json:"guarantee_breaches"`
	// ContractGrossPerHour is the sum of live hourly spreads. Not annualised: that would assume a full
	// time full year assignment nobody agreed to.
	ContractGrossPerHour string `json:"contract_gross_per_hour,omitempty"`
	DirectHireFees       string `json:"direct_hire_fees,omitempty"`
}

const secondsPerDay = 86400

// Analytics computes the report over a window.
//
// The cohort is defined by when a requisition was OPENED, not by when it filled. Cohorting on the fill
// date is the other common way to get this wrong: it drops every order that has not filled yet, which is
// the same survivorship bias the Kaplan Meier estimator exists to correct, reintroduced through the
// window instead of through the average.
func (s *Service) Analytics(ctx context.Context, sinceRaw string) (AnalyticsView, error) {
	if s.db == nil {
		return AnalyticsView{}, ErrNoDatabase
	}
	since, err := optTime("since", sinceRaw)
	if err != nil {
		return AnalyticsView{}, err
	}
	if since == nil {
		t := time.Now().UTC().AddDate(0, 0, -90)
		since = &t
	}
	out := AnalyticsView{
		Since: ts(*since),
		Aging: []AgingBucket{}, Funnel: []FunnelStage{}, Conversion: []Conversion{},
		Sources: []SourceEffect{}, Recruiters: []RecruiterEffort{},
		DeskOutcomes: []DeskOutcome{},
	}
	day := decimal.NewFromInt(secondsPerDay)

	err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		if e := tx.QueryRow(ctx, `SELECT rm_org_tz()`).Scan(&out.TimeZone); e != nil {
			return translatePG(e)
		}

		// One pass over the cohort produces every duration and every outcome count, so the sections of
		// the report cannot disagree about which orders they described.
		//
		// A NULL offer_at or start_on is the censoring signal: that order has not reached the milestone.
		rows, e := tx.Query(ctx, `
			WITH cohort AS (
			  SELECT r.id, r.status, r.opened_at, r.closed_at, r.held_seconds,
			         (SELECT min(o.responded_at) FROM offers o
			            JOIN applications a ON a.organization_id = o.organization_id AND a.id = o.application_id
			           WHERE a.requisition_id = r.id AND o.status = 'accepted') AS offer_at,
			         (SELECT min(p.start_date) FROM placements p
			            JOIN applications a2 ON a2.organization_id = p.organization_id AND a2.id = p.application_id
			           WHERE a2.requisition_id = r.id) AS start_on
			    FROM requisitions r
			   WHERE r.deleted_at IS NULL AND r.opened_at >= $1)
			SELECT status, opened_at, closed_at, held_seconds::text, offer_at, start_on,
			       EXTRACT(EPOCH FROM (now() - opened_at))::text
			  FROM cohort`, *since)
		if e != nil {
			return translatePG(e)
		}
		defer rows.Close()

		var offerObs, startObs, workedObs []Observation
		var openAges []int
		filled, closedUnfilled, cancelled, stillOpen := 0, 0, 0, 0

		for rows.Next() {
			var status, heldStr, aliveStr string
			var openedAt time.Time
			var closedAt, offerAt, startOn *time.Time
			if e := rows.Scan(&status, &openedAt, &closedAt, &heldStr, &offerAt, &startOn, &aliveStr); e != nil {
				return translatePG(e)
			}
			held, _ := decimal.NewFromString(heldStr)
			alive, _ := decimal.NewFromString(aliveStr)
			aliveDays := alive.Div(day)
			heldDays := held.Div(day)

			switch status {
			case reqFilled:
				filled++
			case reqClosed:
				closedUnfilled++
			case reqCancelled:
				cancelled++
			default:
				stillOpen++
				d, _ := aliveDays.Float64()
				openAges = append(openAges, int(d))
				if int(d) > out.OldestOpenDays {
					out.OldestOpenDays = int(d)
				}
			}

			// An order that closed was only observed until it closed. Censoring a closed unfilled order
			// at today instead would claim we are still waiting on a role nobody is working.
			observed := aliveDays
			if closedAt != nil {
				observed = secondsBetween(openedAt, *closedAt).Div(day)
			}

			offerObs = append(offerObs, milestone(openedAt, offerAt, observed, day))
			st := milestone(openedAt, startOn, observed, day)
			startObs = append(startObs, st)
			workedObs = append(workedObs, Observation{Days: nonNegative(st.Days.Sub(heldDays)), Filled: st.Filled})
		}
		if e := rows.Err(); e != nil {
			return e
		}

		out.TimeToOffer = timeToFillFrom(offerObs)
		out.TimeToStart = timeToFillFrom(startObs)
		out.WorkedTimeToStart = timeToFillFrom(workedObs)
		out.FillRate = fillRateFrom(filled, closedUnfilled, cancelled, stillOpen)
		out.Aging = agingFrom(openAges)

		// The funnel over applications to cohort requisitions, from high water marks so it is monotone.
		marks, total := []int{}, 0
		fr, e := tx.Query(ctx, `
			SELECT a.furthest_stage_ord FROM applications a
			  JOIN requisitions r ON r.organization_id = a.organization_id AND r.id = a.requisition_id
			 WHERE r.deleted_at IS NULL AND r.opened_at >= $1`, *since)
		if e != nil {
			return translatePG(e)
		}
		for fr.Next() {
			var m int
			if e := fr.Scan(&m); e != nil {
				fr.Close()
				return translatePG(e)
			}
			marks = append(marks, m)
			total++
		}
		fr.Close()
		if e := fr.Err(); e != nil {
			return e
		}
		out.Funnel = funnelFrom(marks, total)
		out.Conversion = conversionFrom(out.Funnel)

		// Source effectiveness on FIRST touch, the immutable one. A later touch says how a candidate was
		// re engaged, not how the relationship began, and attributing placements to the last touch is how
		// every channel eventually looks like "referral".
		sr, e := tx.Query(ctx, `
			SELECT COALESCE(NULLIF(a.source_first_touch, ''), 'unrecorded') AS src,
			       count(*), count(*) FILTER (WHERE a.stage = 'placed')
			  FROM applications a
			  JOIN requisitions r ON r.organization_id = a.organization_id AND r.id = a.requisition_id
			 WHERE r.deleted_at IS NULL AND r.opened_at >= $1
			 GROUP BY 1 ORDER BY 2 DESC, 1 ASC`, *since)
		if e != nil {
			return translatePG(e)
		}
		for sr.Next() {
			var x SourceEffect
			if e := sr.Scan(&x.Source, &x.Applications, &x.Placed); e != nil {
				sr.Close()
				return translatePG(e)
			}
			if x.Applications < minCohortForRate {
				x.Suppressed = true
			} else {
				x.Rate = rateOf(x.Placed, x.Applications)
			}
			out.Sources = append(out.Sources, x)
		}
		sr.Close()
		if e := sr.Err(); e != nil {
			return e
		}

		// Placement economics, the two shapes kept apart.
		var contractGross, fees *string
		if e := tx.QueryRow(ctx, `
			SELECT count(*) FILTER (WHERE p.status IN ('pending_start','active')),
			       count(*),
			       count(*) FILTER (WHERE p.status IN ('completed','terminated','fell_off')),
			       count(*) FILTER (WHERE p.status = 'fell_off'),
			       count(*) FILTER (WHERE p.status IN ('fell_off','terminated')
			                          AND p.end_date IS NOT NULL
			                          AND p.end_date <= p.start_date + (p.guarantee_days - 1)),
			       sum(p.bill_rate - p.pay_rate) FILTER (
			             WHERE p.employment_type <> 'direct_hire'
			               AND p.status IN ('pending_start','active')
			               AND p.bill_rate IS NOT NULL AND p.pay_rate IS NOT NULL)::text,
			       sum(p.fee_amount) FILTER (WHERE p.employment_type = 'direct_hire')::text
			  FROM placements p
			  JOIN applications a ON a.organization_id = p.organization_id AND a.id = p.application_id
			  JOIN requisitions r ON r.organization_id = a.organization_id AND r.id = a.requisition_id
			 WHERE r.deleted_at IS NULL AND r.opened_at >= $1`, *since).
			Scan(&out.Placements.Live, &out.Placements.Started, &out.Placements.Ended,
				&out.Placements.FellOff, &out.Placements.GuaranteeBreach, &contractGross, &fees); e != nil {
			return translatePG(e)
		}
		out.Placements.ContractGrossPerHour = str(contractGross)
		out.Placements.DirectHireFees = str(fees)

		// Both new sections read on the SAME transaction as everything above, so the report cannot describe
		// one window in its funnel and a different one in its desk outcomes.
		desks, e := s.readDeskOutcomes(ctx, tx, *since)
		if e != nil {
			return e
		}
		out.DeskOutcomes = desks

		rd, e := s.readRedeploymentRate(ctx, tx, *since)
		if e != nil {
			return e
		}
		out.Redeployment = rd
		return nil
	})
	if err != nil {
		return AnalyticsView{}, err
	}

	// Recruiter effort comes from the activity log, which already separates logged from captured effort
	// and is already append only, so it is reused rather than recomputed a second way.
	summary, err := s.ActivitySummary(ctx, ts(*since))
	if err != nil {
		return AnalyticsView{}, err
	}
	out.Recruiters = summary.Recruiters
	return out, nil
}

// milestone turns a reached or unreached milestone into an observation.
//
// Reached is an event; unreached is right censored at however long the order was actually observed. A
// milestone dated before the order opened is a data entry error rather than a negative duration, so it
// is clamped to zero rather than allowed to drag the survival curve below the axis.
func milestone(openedAt time.Time, at *time.Time, observed, day decimal.Decimal) Observation {
	if at == nil {
		return Observation{Days: nonNegative(observed), Filled: false}
	}
	return Observation{Days: nonNegative(secondsBetween(openedAt, *at).Div(day)), Filled: true}
}

func nonNegative(d decimal.Decimal) decimal.Decimal {
	if d.IsNegative() {
		return decimal.Zero
	}
	return d
}