Janady13/missionmatch / internal/modules/matching/merge.go

Source identity

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

Download complete source text ↓347 lines · 14.0 KB
package matching

// CANDIDATE MERGE. Two rows are the same person; the history has to end up on one of them.
//
// A bench gets dirty. The same person applies through the careers page, is imported from an ATS export, and
// is sourced by a recruiter, and now there are three of them with a third of the history each. Dedup already
// existed at IMPORT time; nothing existed for the duplicates that arrive afterwards, which is most of them.
//
// THREE PROPERTIES MAKE THIS SAFE, and each exists because the obvious implementation gets it wrong.
//
//  1. THE TABLES ARE DISCOVERED FROM THE CATALOG. Fifteen foreign keys across twelve migrations point at
//     candidates today and the next migration will add another. A hand maintained list in Go is a list that
//     is wrong the first time somebody forgets it, and the failure is silent: the merge reports success and
//     a candidate's interviews stay attached to a tombstone nobody can see.
//
//  2. EACH TABLE MOVES INSIDE ITS OWN SAVEPOINT. Repointing rows can violate a unique constraint whenever
//     BOTH candidates already had the same thing, which for two records of one person is the common case
//     rather than the exception. Without a savepoint the first collision aborts the entire transaction and
//     the merge fails on exactly the duplicates it exists to fix.
//
//  3. THE LOSER IS A TOMBSTONE, NEVER A DELETE. This is an Automated Employment Decision Tool: a scoring
//     decision made in March about that candidate must be reconstructable in June, and deleting the row
//     destroys the subject of the decision.

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"strings"
	"time"

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

	"github.com/missionworkforce/missionmatch/internal/platform/audit"
)

var (
	ErrMergeSame     = errors.New("matching: a candidate cannot be merged into itself")
	ErrMergeUnknown  = errors.New("matching: one of those candidates does not exist in this organization")
	ErrAlreadyMerged = errors.New("matching: that candidate has already been merged into another record")
)

// MergeResult reports what moved and what could not, per table. "moved 41 rows" without saying from where is
// an answer nobody can check; "dropped 3" without saying which table is worse.
type MergeResult struct {
	KeptID   string         `json:"kept_id"`
	MergedID string         `json:"merged_id"`
	Moved    map[string]int `json:"moved"`
	Dropped  map[string]int `json:"dropped,omitempty"`
	// Retained names tables whose rows DELIBERATELY stayed on the tombstone. It is distinct from Dropped,
	// and the distinction is the point: dropped means a duplicate row was removed, retained means the row is
	// a fact about the merged record that must not be rewritten. AEDT compliance notices are the case that
	// forced this to exist.
	Retained []string `json:"retained_on_the_merged_record,omitempty"`
	Note     string   `json:"note,omitempty"`
}

// DuplicateCandidate is one suggested pair.
type DuplicateCandidate struct {
	AID    string `json:"a_id"`
	AName  string `json:"a_name"`
	BID    string `json:"b_id"`
	BName  string `json:"b_name"`
	Reason string `json:"reason"`
	// Confidence is deliberately coarse. A number to two decimal places on a duplicate suggestion implies a
	// calibration this does not have, and a recruiter treats 0.87 as a fact rather than a hint.
	Strength string `json:"strength"`
}

// FindDuplicates suggests pairs. IT ONLY EVER SUGGESTS: nothing here merges anything, because merging two
// people who share a common name is a mistake that cannot be undone by anybody who was not watching.
func (s *Service) FindDuplicates(ctx context.Context, limit int) ([]DuplicateCandidate, error) {
	if s.db == nil {
		return nil, ErrNoDatabase
	}
	if limit <= 0 || limit > 200 {
		limit = 50
	}
	out := make([]DuplicateCandidate, 0, 32)
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		// EXACT EMAIL is the strong signal and comes first. Two records with the same address are the same
		// person in every practical case, and it is the only rule here confident enough to be called strong.
		rows, e := tx.Query(ctx, `
			SELECT a.id::text, a.full_name, b.id::text, b.full_name, 'same email address', 'strong'
			FROM candidates a JOIN candidates b
			  ON lower(a.email) = lower(b.email) AND a.id < b.id
			WHERE a.deleted_at IS NULL AND b.deleted_at IS NULL
			  AND a.email IS NOT NULL AND btrim(a.email) <> ''
			UNION ALL
			-- NAME PLUS LOCATION is a HINT, not a finding. Two cleared analysts called James Smith in the
			-- same metro area is an ordinary thing, so this is offered for a human to look at and is never
			-- described as a match.
			SELECT a.id::text, a.full_name, b.id::text, b.full_name, 'same name and location', 'weak'
			FROM candidates a JOIN candidates b
			  ON lower(btrim(a.full_name)) = lower(btrim(b.full_name))
			 AND lower(coalesce(a.location,'')) = lower(coalesce(b.location,''))
			 AND a.id < b.id
			WHERE a.deleted_at IS NULL AND b.deleted_at IS NULL
			  AND btrim(a.full_name) <> ''
			  AND (a.email IS NULL OR b.email IS NULL OR lower(a.email) <> lower(b.email))
			LIMIT $1`, limit)
		if e != nil {
			return e
		}
		defer rows.Close()
		for rows.Next() {
			var d DuplicateCandidate
			if e := rows.Scan(&d.AID, &d.AName, &d.BID, &d.BName, &d.Reason, &d.Strength); e != nil {
				return e
			}
			out = append(out, d)
		}
		return rows.Err()
	})
	return out, err
}

// MergeCandidates moves everything from mergedID onto keptID and tombstones mergedID.
func (s *Service) MergeCandidates(ctx context.Context, keptID, mergedID uuid.UUID, reason, ip, ua string) (MergeResult, error) {
	res := MergeResult{
		KeptID: keptID.String(), MergedID: mergedID.String(),
		Moved: map[string]int{}, Dropped: map[string]int{},
	}
	if s.db == nil {
		return res, ErrNoDatabase
	}
	if keptID == mergedID {
		return res, ErrMergeSame
	}
	actor := actorUUIDPtr(ctx)
	if len(reason) > 500 {
		reason = reason[:500]
	}

	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		// BOTH must exist, be live, and not already be tombstones. Merging into a tombstone would bury the
		// history one level deeper every time somebody tried.
		// The row is looked up WITHOUT the deleted_at filter and the two conditions are then separated.
		// Filtering it out first made an already merged record report "no such candidate", which is both
		// wrong and actively misleading: it tells somebody the record does not exist when in fact it is
		// exactly where they left it, as a tombstone.
		for _, id := range []uuid.UUID{keptID, mergedID} {
			var merged *uuid.UUID
			var deleted *time.Time
			e := tx.QueryRow(ctx, `SELECT merged_into_id, deleted_at FROM candidates WHERE id = $1`, id).
				Scan(&merged, &deleted)
			if errors.Is(e, pgx.ErrNoRows) {
				return ErrMergeUnknown
			}
			if e != nil {
				return e
			}
			if merged != nil {
				return ErrAlreadyMerged
			}
			if deleted != nil {
				// Soft deleted for some other reason. Not a tombstone, but still not something to merge.
				return ErrMergeUnknown
			}
		}

		// The tables are read from the catalog every time, so a table added by a later migration is
		// repointed without anybody remembering to come back here.
		type ref struct{ table, column string }
		refs := make([]ref, 0, 24)
		rows, e := tx.Query(ctx, `SELECT table_name, column_name FROM mm_candidate_referencing_tables()`)
		if e != nil {
			return fmt.Errorf("matching: read referencing tables: %w", e)
		}
		for rows.Next() {
			var r ref
			if e := rows.Scan(&r.table, &r.column); e != nil {
				rows.Close()
				return e
			}
			// The identifiers come from the catalog and can only be real table and column names, but they
			// are interpolated into SQL, so they are checked anyway. A catalog this trusted is exactly where
			// nobody looks when something goes wrong.
			if !bareIdentifier(r.table) || !bareIdentifier(r.column) {
				rows.Close()
				return fmt.Errorf("matching: refusing to repoint through an unexpected identifier %q.%q", r.table, r.column)
			}
			// The merge LEDGER is never rewritten. Repointing a past merge's kept_id or merged_id would
			// rewrite the record of what was merged, which is the one table that has to stay true about the
			// operation itself.
			if r.table == "candidate_merges" {
				continue
			}
			refs = append(refs, r)
		}
		rows.Close()
		if rows.Err() != nil {
			return rows.Err()
		}

		for _, r := range refs {
			// A SAVEPOINT PER TABLE. Repointing collides whenever both candidates already had the same thing,
			// which for two records of one person is the common case, and without this the first collision
			// aborts the whole merge.
			sp, e := tx.Begin(ctx)
			if e != nil {
				return e
			}
			tag, e := sp.Exec(ctx,
				fmt.Sprintf(`UPDATE %s SET %s = $1 WHERE %s = $2`, r.table, r.column, r.column),
				keptID, mergedID)
			if e == nil {
				if n := int(tag.RowsAffected()); n > 0 {
					res.Moved[r.table] = n
				}
				if e := sp.Commit(ctx); e != nil {
					return e
				}
				continue
			}
			_ = sp.Rollback(ctx)
			if isPermissionDenied(e) {
				// THE ROW STAYS WHERE IT IS, AND THAT IS CORRECT RATHER THAN A LIMITATION.
				//
				// The application role has no UPDATE on aedt_candidate_notices, deliberately: a notice sent
				// to a candidate under an Automated Employment Decision Tool disclosure rule is a FACT ABOUT
				// THAT RECORD, and repointing it at another person would falsify a compliance record to make
				// a tidy merge. The merge records that it left them and carries on. The tombstone still
				// resolves through merged_into_id, so nothing becomes unreachable.
				res.Retained = append(res.Retained, r.table)
				continue
			}
			if !isUniqueViolation(e) {
				// The cause is LOGGED. A merge that fails with a generic message is a merge nobody can fix,
				// and this walks eighteen tables whose failure modes differ.
				log.Printf("matching: repoint %s.%s: %v", r.table, r.column, e)
				return fmt.Errorf("matching: repoint %s: %w", r.table, e)
			}
			// BOTH SIDES ALREADY HAD IT. The loser's row is dropped rather than moved, and the count is
			// RECORDED, because a merge that silently discarded rows is indistinguishable from one that
			// moved them and the difference matters six months later.
			sp2, e2 := tx.Begin(ctx)
			if e2 != nil {
				return e2
			}
			tag2, e2 := sp2.Exec(ctx,
				fmt.Sprintf(`DELETE FROM %s WHERE %s = $1`, r.table, r.column), mergedID)
			if e2 != nil {
				_ = sp2.Rollback(ctx)
				// The table has no DELETE grant, which is true of most of this schema by design. The rows
				// stay on the tombstone, which still resolves through merged_into_id, and that is recorded
				// rather than hidden.
				res.Dropped[r.table] = -1
				continue
			}
			res.Dropped[r.table] = int(tag2.RowsAffected())
			if e := sp2.Commit(ctx); e != nil {
				return e
			}
		}

		// FIELDS ARE FILLED IN, NEVER OVERWRITTEN. The kept record's own values win; the loser only
		// contributes where the winner is empty. Overwriting would let a merge quietly replace a verified
		// clearance with an unverified one, which is the single worst outcome this operation could have.
		if _, e := tx.Exec(ctx, `
			UPDATE candidates k SET
			  email          = coalesce(nullif(btrim(k.email),''), m.email),
			  headline       = coalesce(nullif(btrim(k.headline),''), m.headline),
			  location       = coalesce(nullif(btrim(k.location),''), m.location),
			  owner_user_id  = coalesce(k.owner_user_id, m.owner_user_id),
			  years_experience = greatest(k.years_experience, m.years_experience),
			  updated_at     = now()
			FROM candidates m
			WHERE k.id = $1 AND m.id = $2`, keptID, mergedID); e != nil {
			return e
		}

		if _, e := tx.Exec(ctx, `
			UPDATE candidates SET merged_into_id = $2, deleted_at = now(), updated_at = now()
			WHERE id = $1`, mergedID, keptID); e != nil {
			return e
		}

		moved, _ := json.Marshal(res.Moved)
		dropped, _ := json.Marshal(res.Dropped)
		if _, e := tx.Exec(ctx, `
			INSERT INTO candidate_merges (organization_id, kept_id, merged_id, moved, dropped, reason, merged_by)
			VALUES (NULLIF(current_setting('app.current_org_id', true), '')::uuid, $1,$2,$3,$4,$5,$6)`,
			keptID, mergedID, moved, dropped, reason, actor); e != nil {
			return e
		}
		if s.aud == nil {
			return nil
		}
		body, _ := json.Marshal(map[string]any{
			"kept_id": keptID.String(), "merged_id": mergedID.String(),
			"moved": res.Moved, "dropped": res.Dropped, "reason": reason,
		})
		return s.aud.OnTx(ctx, tx, audit.Event{
			Action: "candidate.merged", ActorUserID: actor, EntityType: "candidate",
			EntityID: keptID.String(), NewValues: body, IPAddress: ip, UserAgent: ua,
		})
	})
	if err != nil {
		return MergeResult{KeptID: res.KeptID, MergedID: res.MergedID}, err
	}
	res.Note = "the merged record is retained as a tombstone and still resolves to the kept one, " +
		"because a scoring decision about it has to stay reconstructable"
	if len(res.Retained) > 0 {
		res.Note += ". Compliance records naming the merged candidate were deliberately left on it, " +
			"because a notice sent to that person is a fact about them and must not be reattributed"
	}
	if len(res.Dropped) == 0 {
		res.Dropped = nil
	}
	return res, nil
}

// bareIdentifier is the guard on names that reach SQL. They come from the catalog and cannot realistically be
// anything else, which is exactly why it is checked: a value nobody doubts is a value nobody examines.
func bareIdentifier(s string) bool {
	if s == "" || len(s) > 63 {
		return false
	}
	for i, r := range s {
		switch {
		case r >= 'a' && r <= 'z':
		case r == '_':
		case (r >= '0' && r <= '9') && i > 0:
		default:
			return false
		}
	}
	return true
}

// isPermissionDenied recognises a table the application role may not update. That is a DESIGN DECISION
// somewhere else in this schema, not an accident, so the merge respects it rather than failing on it.
func isPermissionDenied(err error) bool {
	m := err.Error()
	return strings.Contains(m, "SQLSTATE 42501") || strings.Contains(m, "permission denied")
}

func isUniqueViolation(err error) bool {
	m := err.Error()
	return strings.Contains(m, "duplicate key") || strings.Contains(m, "unique constraint") ||
		strings.Contains(m, "SQLSTATE 23505")
}