Janady13/missionmatch / internal/modules/opportunities/scope_splitmerge.go

Source identity

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

Download complete source text ↓289 lines · 10.9 KB
package opportunities

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/missionworkforce/missionmatch/internal/platform/audit"
	"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)

// SPLIT AND MERGE, directive Section 6 screen 4 and Phase 2 item 4, which name six review actions. Four
// existed. These are the other two.
//
// THE CITATION SURVIVES BOTH OPERATIONS, and that is the whole design constraint. A split child inherits the
// parent's source document, page, section label and quote hash, so it is grounded in exactly the span the
// parent was grounded in. A merge keeps the surviving item's citation. Neither operation can mint a machine
// produced item with no source, and ck_scope_items_cited remains the backstop if either ever tries.
//
// NOTHING IS DELETED, EVER. A split parent is retired to status 'split' and a merged item to 'merged', both
// keeping every column they had, with parent_scope_item_id recording the lineage. The review queue filters
// them out; the audit trail and the row itself remain. This mirrors the accounts rule elsewhere in the
// platform, where deactivation preserves history and deletion is not offered.

// SplitPart is one child of a split. ScopeType is optional: empty means inherit the parent's classification,
// because making the reviewer restate it on every child is an invitation to get it wrong.
type SplitPart struct {
	ScopeType   string `json:"scope_type"`
	Title       string `json:"title"`
	Description string `json:"description"`
}

// maxSplitParts bounds one split. An unbounded split is a way to flood the review queue from a single item,
// and no real clause divides into fifty tasks.
const maxSplitParts = 50

func validateSplit(parts []SplitPart) error {
	if len(parts) < 2 {
		return fmt.Errorf("%w: a split needs at least two parts, %d given. Splitting into one is an edit",
			ErrInvalidInput, len(parts))
	}
	if len(parts) > maxSplitParts {
		return fmt.Errorf("%w: a split is limited to %d parts, %d given", ErrInvalidInput, maxSplitParts, len(parts))
	}
	for i, p := range parts {
		if strings.TrimSpace(p.Title) == "" {
			return fmt.Errorf("%w: part %d has no title", ErrInvalidInput, i+1)
		}
		if st := strings.TrimSpace(p.ScopeType); st != "" && !scopeTypes[st] {
			return fmt.Errorf("%w: part %d scope_type %q is not valid", ErrInvalidInput, i+1, st)
		}
	}
	return nil
}

func validateMerge(primaryID string, absorbIDs []string, title string) error {
	pid, err := uuid.Parse(strings.TrimSpace(primaryID))
	if err != nil {
		return fmt.Errorf("%w: item id must be a uuid", ErrInvalidInput)
	}
	if len(absorbIDs) == 0 {
		return fmt.Errorf("%w: a merge needs at least one item to absorb, otherwise it is an edit", ErrInvalidInput)
	}
	if strings.TrimSpace(title) == "" {
		return fmt.Errorf("%w: the merged item needs a title saying what it now covers", ErrInvalidInput)
	}
	seen := map[uuid.UUID]bool{}
	for _, raw := range absorbIDs {
		id, pErr := uuid.Parse(strings.TrimSpace(raw))
		if pErr != nil {
			return fmt.Errorf("%w: %q is not a uuid", ErrInvalidInput, raw)
		}
		if id == pid {
			return fmt.Errorf("%w: an item cannot absorb itself", ErrInvalidInput)
		}
		if seen[id] {
			return fmt.Errorf("%w: %s appears twice in the absorb list", ErrInvalidInput, id)
		}
		seen[id] = true
	}
	return nil
}

// SplitScopeItem divides one item into two or more, each carrying the parent's citation, and retires the
// parent. It is ONE transaction: a half applied split would leave the parent retired with no children, which
// silently loses scope a reviewer was looking at.
func (s *Service) SplitScopeItem(ctx context.Context, itemID string, parts []SplitPart) ([]ScopeItem, error) {
	if !s.db.Ready() {
		return nil, ErrNoDatabase
	}
	iid, err := uuid.Parse(strings.TrimSpace(itemID))
	if err != nil {
		return nil, fmt.Errorf("%w: item id must be a uuid", ErrInvalidInput)
	}
	if vErr := validateSplit(parts); vErr != nil {
		return nil, vErr
	}
	actor := actorPtr(ctx)
	if actor == nil {
		return nil, tenant.ErrNoTenant
	}

	out := make([]ScopeItem, 0, len(parts))
	err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		// FOR UPDATE, the same discipline promotion uses. Two reviewers splitting the same item concurrently
		// would otherwise each retire it and each write their own children, doubling the scope.
		var parentType, docID, sectionLabel, quoteHash, method, status string
		var page int
		var conf float64
		var oppID string
		row := tx.QueryRow(ctx, `
			SELECT opportunity_id::text, scope_type, COALESCE(source_document_id::text,''),
			       COALESCE(source_page_number,0), source_section_label, source_quote_hash,
			       extraction_method, extraction_confidence, review_status
			FROM opportunity_scope_items WHERE id = $1 FOR UPDATE`, iid)
		if sErr := row.Scan(&oppID, &parentType, &docID, &page, &sectionLabel, &quoteHash,
			&method, &conf, &status); sErr != nil {
			return sErr
		}
		// An already retired item must not be split again: its children would be orphaned from the queue.
		if status == "split" || status == "merged" || status == "rejected" {
			return fmt.Errorf("%w: this item is already %s and cannot be split", ErrInvalidInput, status)
		}

		for i, p := range parts {
			st := strings.TrimSpace(p.ScopeType)
			if st == "" {
				st = parentType
			}
			var docArg any
			if docID != "" {
				docArg = docID
			}
			var pageArg any
			if page > 0 {
				pageArg = page
			}
			child := tx.QueryRow(ctx, `
				INSERT INTO opportunity_scope_items (
				  organization_id, opportunity_id, scope_type, title, description,
				  source_document_id, source_page_number, source_section_label, source_quote_hash,
				  extraction_method, extraction_confidence, review_status,
				  parent_scope_item_id, reviewed_by_user_id, reviewed_at, display_order)
				VALUES (NULLIF(current_setting('app.current_org_id', true), '')::uuid,
				        $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, 'edited', $11, $12, now(), $13)
				RETURNING `+scopeCols,
				oppID, st, strings.TrimSpace(p.Title), p.Description,
				docArg, pageArg, sectionLabel, quoteHash, method, conf, iid, actor, i)
			it, sErr := scanScope(child)
			if sErr != nil {
				return sErr
			}
			out = append(out, it)
		}

		if _, uErr := tx.Exec(ctx, `
			UPDATE opportunity_scope_items
			SET review_status = 'split', reviewed_by_user_id = $1, reviewed_at = now(), updated_at = now()
			WHERE id = $2`, actor, iid); uErr != nil {
			return uErr
		}

		ids := make([]string, 0, len(out))
		for _, it := range out {
			ids = append(ids, it.ID)
		}
		nv, _ := json.Marshal(map[string]any{"parent_scope_item_id": iid.String(), "child_ids": ids,
			"parts": len(out)})
		return s.aud.OnTx(ctx, tx, audit.Event{
			Action: "opportunity.scope.split", ActorUserID: actor,
			EntityType: "opportunity_scope_item", EntityID: iid.String(), NewValues: nv,
		})
	})
	if err != nil {
		return nil, err
	}
	return out, nil
}

// MergeScopeItems folds one or more items into a surviving item, which keeps its own citation and takes the
// combined text. The absorbed items are retired to 'merged' and point at the survivor.
func (s *Service) MergeScopeItems(ctx context.Context, primaryID string, absorbIDs []string,
	title, description, scopeType string) (ScopeItem, error) {
	if !s.db.Ready() {
		return ScopeItem{}, ErrNoDatabase
	}
	if vErr := validateMerge(primaryID, absorbIDs, title); vErr != nil {
		return ScopeItem{}, vErr
	}
	if st := strings.TrimSpace(scopeType); st != "" && !scopeTypes[st] {
		return ScopeItem{}, fmt.Errorf("%w: scope_type %q is not valid", ErrInvalidInput, st)
	}
	pid, _ := uuid.Parse(strings.TrimSpace(primaryID))
	actor := actorPtr(ctx)
	if actor == nil {
		return ScopeItem{}, tenant.ErrNoTenant
	}

	var out ScopeItem
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		// Lock the survivor first, then every absorbed row, always in a stable order so two merges touching
		// the same pair cannot deadlock each other.
		var status string
		if sErr := tx.QueryRow(ctx,
			`SELECT review_status FROM opportunity_scope_items WHERE id = $1 FOR UPDATE`, pid).
			Scan(&status); sErr != nil {
			return sErr
		}
		if status == "split" || status == "merged" || status == "rejected" {
			return fmt.Errorf("%w: the surviving item is already %s", ErrInvalidInput, status)
		}

		ids := make([]uuid.UUID, 0, len(absorbIDs))
		for _, raw := range absorbIDs {
			id, _ := uuid.Parse(strings.TrimSpace(raw))
			ids = append(ids, id)
		}
		// EVERY ABSORBED ROW MUST EXIST AND BELONG TO THIS TENANT. RLS handles the tenant half; the count
		// check catches a caller passing an id from another opportunity, which would silently retire an item
		// a different reviewer is working on.
		//
		// THE ROWS ARE SELECTED AND COUNTED IN GO RATHER THAN BY count(*). Postgres refuses FOR UPDATE
		// alongside an aggregate, "FOR UPDATE is not allowed with aggregate functions", so the first version of
		// this query was invalid SQL that no pure test could have caught. Locking the rows is the point, so the
		// lock stays and the counting moves.
		lockRows, lErr := tx.Query(ctx,
			`SELECT id FROM opportunity_scope_items
			  WHERE id = ANY($1) AND review_status NOT IN ('split','merged','rejected')
			  ORDER BY id
			  FOR UPDATE`, ids)
		if lErr != nil {
			return lErr
		}
		found := 0
		for lockRows.Next() {
			found++
		}
		lockRows.Close()
		if rErr := lockRows.Err(); rErr != nil {
			return rErr
		}
		if found != len(ids) {
			return fmt.Errorf("%w: %d of %d items to absorb are missing or already retired",
				ErrInvalidInput, len(ids)-found, len(ids))
		}

		sets := []string{"title = $1", "review_status = 'edited'", "reviewed_by_user_id = $2",
			"reviewed_at = now()", "updated_at = now()"}
		args := []any{strings.TrimSpace(title), actor}
		if strings.TrimSpace(description) != "" {
			args = append(args, description)
			sets = append(sets, fmt.Sprintf("description = $%d", len(args)))
		}
		if st := strings.TrimSpace(scopeType); st != "" {
			args = append(args, st)
			sets = append(sets, fmt.Sprintf("scope_type = $%d", len(args)))
		}
		args = append(args, pid)
		survivor := tx.QueryRow(ctx, `UPDATE opportunity_scope_items SET `+strings.Join(sets, ", ")+
			fmt.Sprintf(" WHERE id = $%d RETURNING ", len(args))+scopeCols, args...)
		var sErr error
		if out, sErr = scanScope(survivor); sErr != nil {
			return sErr
		}

		if _, uErr := tx.Exec(ctx, `
			UPDATE opportunity_scope_items
			SET review_status = 'merged', parent_scope_item_id = $1, reviewed_by_user_id = $2,
			    reviewed_at = now(), updated_at = now()
			WHERE id = ANY($3)`, pid, actor, ids); uErr != nil {
			return uErr
		}

		nv, _ := json.Marshal(map[string]any{"surviving_scope_item_id": out.ID, "absorbed": absorbIDs,
			"absorbed_count": len(ids)})
		return s.aud.OnTx(ctx, tx, audit.Event{
			Action: "opportunity.scope.merged", ActorUserID: actor,
			EntityType: "opportunity_scope_item", EntityID: out.ID, NewValues: nv,
		})
	})
	if err != nil {
		return ScopeItem{}, err
	}
	return out, nil
}