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

Source identity

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

Download complete source text ↓334 lines · 11.2 KB
package opportunities

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

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

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

// The closed vocabularies for the capture path, mirroring the CHECK constraints in migration 0070 so a bad
// value is refused with a 400 before it reaches the database rather than as a constraint violation.
var (
	validSourceTypes = map[string]bool{"sam_gov": true, "agency_forecast": true, "customer": true, "partner": true, "internal": true, "spreadsheet": true, "manual": true}
	validStatuses    = map[string]bool{"identified": true, "qualifying": true, "qualified": true, "pursuing": true, "no_bid": true, "submitted": true, "won": true, "lost": true, "cancelled": true, "archived": true}
	validPrimeOrSub  = map[string]bool{"": true, "prime": true, "sub": true}
)

// trackedCols is the shared projection for reads and RETURNING, so every code path builds the view from the
// same columns. capture_owner_user_id and id are cast to text so a nullable uuid scans into a string.
const trackedCols = `id::text, source_type, title, agency_name, office_name, solicitation_number,
	notice_type, response_deadline, prime_or_sub, status, capture_owner_user_id::text, naics_code,
	psc_code, set_aside_code, place_of_performance, incumbent_name, source_url, created_at, updated_at`

// CreateTracked inserts a tenant tracked opportunity under the caller's org. organization_id is set from the
// transaction GUC, never from Go, so it matches the RLS WITH CHECK policy, and created_by_user_id is the
// authenticated actor. The write is audited in the same transaction.
func (s *Service) CreateTracked(ctx context.Context, in TrackedOppInput) (TrackedOpportunity, error) {
	if !s.db.Ready() {
		return TrackedOpportunity{}, ErrNoDatabase
	}
	in.SourceType = strings.TrimSpace(in.SourceType)
	in.Title = strings.TrimSpace(in.Title)
	in.PrimeOrSub = strings.TrimSpace(in.PrimeOrSub)
	if in.Title == "" {
		return TrackedOpportunity{}, fmt.Errorf("%w: title is required", ErrInvalidInput)
	}
	if !validSourceTypes[in.SourceType] {
		return TrackedOpportunity{}, fmt.Errorf("%w: source_type %q is not a valid source", ErrInvalidInput, in.SourceType)
	}
	status := strings.TrimSpace(in.Status)
	if status == "" {
		status = "identified"
	}
	if !validStatuses[status] {
		return TrackedOpportunity{}, fmt.Errorf("%w: status %q is not a valid status", ErrInvalidInput, status)
	}
	if !validPrimeOrSub[in.PrimeOrSub] {
		return TrackedOpportunity{}, fmt.Errorf("%w: prime_or_sub %q is not valid", ErrInvalidInput, in.PrimeOrSub)
	}
	deadline, err := parseDeadline(in.ResponseDeadline)
	if err != nil {
		return TrackedOpportunity{}, err
	}
	owner, err := parseUUIDPtr(in.CaptureOwnerUserID)
	if err != nil {
		return TrackedOpportunity{}, err
	}
	actor := actorPtr(ctx)
	if actor == nil {
		return TrackedOpportunity{}, tenant.ErrNoTenant
	}

	var out TrackedOpportunity
	err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		row := tx.QueryRow(ctx, `
			INSERT INTO tracked_opportunities (
			  organization_id, source_type, title, agency_name, office_name, solicitation_number, notice_type,
			  response_deadline, prime_or_sub, status, capture_owner_user_id, naics_code, psc_code,
			  set_aside_code, place_of_performance, incumbent_name, source_url, created_by_user_id)
			VALUES (
			  NULLIF(current_setting('app.current_org_id', true), '')::uuid,
			  $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17)
			RETURNING `+trackedCols,
			in.SourceType, in.Title, in.AgencyName, in.OfficeName, in.SolicitationNumber, in.NoticeType,
			deadline, in.PrimeOrSub, status, owner, in.NaicsCode, in.PscCode, in.SetAsideCode,
			in.PlaceOfPerformance, in.IncumbentName, in.SourceURL, actor)
		var scanErr error
		if out, scanErr = scanTracked(row); scanErr != nil {
			return scanErr
		}
		nv, _ := json.Marshal(map[string]any{"id": out.ID, "source_type": out.SourceType, "status": out.Status})
		return s.aud.OnTx(ctx, tx, audit.Event{
			Action: "opportunity.created", ActorUserID: actor,
			EntityType: "tracked_opportunity", EntityID: out.ID, NewValues: nv,
		})
	})
	if err != nil {
		return TrackedOpportunity{}, err
	}
	return out, nil
}

// ListTracked returns the caller's tracked opportunities, newest first, optionally filtered by status. RLS
// scopes the read to the caller's org, so no organization_id predicate is written in Go.
func (s *Service) ListTracked(ctx context.Context, statusFilter string, limit int) ([]TrackedOpportunity, error) {
	if !s.db.Ready() {
		return nil, ErrNoDatabase
	}
	if limit <= 0 {
		limit = defaultLimit
	}
	if limit > maxLimit {
		limit = maxLimit
	}
	out := make([]TrackedOpportunity, 0)
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		var rows pgx.Rows
		var qErr error
		if sf := strings.TrimSpace(statusFilter); sf != "" {
			rows, qErr = tx.Query(ctx, `SELECT `+trackedCols+` FROM tracked_opportunities WHERE status = $1 ORDER BY created_at DESC LIMIT $2`, sf, limit)
		} else {
			rows, qErr = tx.Query(ctx, `SELECT `+trackedCols+` FROM tracked_opportunities ORDER BY created_at DESC LIMIT $1`, limit)
		}
		if qErr != nil {
			return qErr
		}
		defer rows.Close()
		for rows.Next() {
			t, scanErr := scanTracked(rows)
			if scanErr != nil {
				return scanErr
			}
			out = append(out, t)
		}
		return rows.Err()
	})
	if err != nil {
		return nil, err
	}
	return out, nil
}

// GetTracked returns one tracked opportunity by id under the caller's org. A missing row (or one owned by
// another org, which RLS hides) surfaces pgx.ErrNoRows for the handler to map to 404.
func (s *Service) GetTracked(ctx context.Context, id string) (TrackedOpportunity, error) {
	if !s.db.Ready() {
		return TrackedOpportunity{}, ErrNoDatabase
	}
	oid, err := uuid.Parse(strings.TrimSpace(id))
	if err != nil {
		return TrackedOpportunity{}, fmt.Errorf("%w: id must be a uuid", ErrInvalidInput)
	}
	var out TrackedOpportunity
	err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		row := tx.QueryRow(ctx, `SELECT `+trackedCols+` FROM tracked_opportunities WHERE id = $1`, oid)
		var scanErr error
		out, scanErr = scanTracked(row)
		return scanErr
	})
	if err != nil {
		return TrackedOpportunity{}, err
	}
	return out, nil
}

// PatchTracked applies a partial update. Only the non nil fields change; a request with no fields is a read.
// updated_at is bumped whenever a column changes. The write is audited.
func (s *Service) PatchTracked(ctx context.Context, id string, p TrackedOppPatch) (TrackedOpportunity, error) {
	if !s.db.Ready() {
		return TrackedOpportunity{}, ErrNoDatabase
	}
	oid, err := uuid.Parse(strings.TrimSpace(id))
	if err != nil {
		return TrackedOpportunity{}, fmt.Errorf("%w: id must be a uuid", ErrInvalidInput)
	}
	actor := actorPtr(ctx)
	if actor == nil {
		return TrackedOpportunity{}, tenant.ErrNoTenant
	}

	sets := make([]string, 0, 12)
	args := make([]any, 0, 12)
	add := func(col string, val any) {
		args = append(args, val)
		sets = append(sets, fmt.Sprintf("%s = $%d", col, len(args)))
	}
	if p.Title != nil {
		add("title", strings.TrimSpace(*p.Title))
	}
	if p.AgencyName != nil {
		add("agency_name", *p.AgencyName)
	}
	if p.OfficeName != nil {
		add("office_name", *p.OfficeName)
	}
	if p.SolicitationNumber != nil {
		add("solicitation_number", *p.SolicitationNumber)
	}
	if p.NoticeType != nil {
		add("notice_type", *p.NoticeType)
	}
	if p.NaicsCode != nil {
		add("naics_code", *p.NaicsCode)
	}
	if p.PscCode != nil {
		add("psc_code", *p.PscCode)
	}
	if p.SetAsideCode != nil {
		add("set_aside_code", *p.SetAsideCode)
	}
	if p.PlaceOfPerformance != nil {
		add("place_of_performance", *p.PlaceOfPerformance)
	}
	if p.IncumbentName != nil {
		add("incumbent_name", *p.IncumbentName)
	}
	if p.SourceURL != nil {
		add("source_url", *p.SourceURL)
	}
	if p.Status != nil {
		v := strings.TrimSpace(*p.Status)
		if !validStatuses[v] {
			return TrackedOpportunity{}, fmt.Errorf("%w: status %q is not a valid status", ErrInvalidInput, v)
		}
		add("status", v)
	}
	if p.PrimeOrSub != nil {
		v := strings.TrimSpace(*p.PrimeOrSub)
		if !validPrimeOrSub[v] {
			return TrackedOpportunity{}, fmt.Errorf("%w: prime_or_sub %q is not valid", ErrInvalidInput, v)
		}
		add("prime_or_sub", v)
	}
	if p.ResponseDeadline != nil {
		d, perr := parseDeadline(*p.ResponseDeadline)
		if perr != nil {
			return TrackedOpportunity{}, perr
		}
		add("response_deadline", d)
	}
	if p.CaptureOwnerUserID != nil {
		o, perr := parseUUIDPtr(*p.CaptureOwnerUserID)
		if perr != nil {
			return TrackedOpportunity{}, perr
		}
		add("capture_owner_user_id", o)
	}
	if len(sets) == 0 {
		return s.GetTracked(ctx, id)
	}
	sets = append(sets, "updated_at = now()")
	args = append(args, oid)
	query := `UPDATE tracked_opportunities SET ` + strings.Join(sets, ", ") +
		fmt.Sprintf(" WHERE id = $%d RETURNING ", len(args)) + trackedCols

	var out TrackedOpportunity
	err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		row := tx.QueryRow(ctx, query, args...)
		var scanErr error
		if out, scanErr = scanTracked(row); scanErr != nil {
			return scanErr
		}
		nv, _ := json.Marshal(map[string]any{"id": out.ID, "status": out.Status})
		return s.aud.OnTx(ctx, tx, audit.Event{
			Action: "opportunity.updated", ActorUserID: actor,
			EntityType: "tracked_opportunity", EntityID: out.ID, NewValues: nv,
		})
	})
	if err != nil {
		return TrackedOpportunity{}, err
	}
	return out, nil
}

// scanTracked builds the view from the shared trackedCols projection, formatting the nullable deadline and
// owner into their string view fields.
func scanTracked(row pgx.Row) (TrackedOpportunity, error) {
	var t TrackedOpportunity
	var respDeadline *time.Time
	var captureOwner *string
	var createdAt, updatedAt time.Time
	if err := row.Scan(&t.ID, &t.SourceType, &t.Title, &t.AgencyName, &t.OfficeName, &t.SolicitationNumber,
		&t.NoticeType, &respDeadline, &t.PrimeOrSub, &t.Status, &captureOwner, &t.NaicsCode,
		&t.PscCode, &t.SetAsideCode, &t.PlaceOfPerformance, &t.IncumbentName, &t.SourceURL,
		&createdAt, &updatedAt); err != nil {
		return TrackedOpportunity{}, err
	}
	if respDeadline != nil {
		t.ResponseDeadline = respDeadline.Format(time.RFC3339)
	}
	if captureOwner != nil {
		t.CaptureOwnerUserID = *captureOwner
	}
	t.CreatedAt = createdAt.Format(time.RFC3339)
	t.UpdatedAt = updatedAt.Format(time.RFC3339)
	return t, nil
}

// actorPtr reads the authenticated user id from context for created_by and the audit actor.
func actorPtr(ctx context.Context) *uuid.UUID {
	p, ok := tenant.UserFromContext(ctx)
	if !ok {
		return nil
	}
	id, err := uuid.Parse(p.UserID)
	if err != nil {
		return nil
	}
	return &id
}

func parseDeadline(s string) (*time.Time, error) {
	s = strings.TrimSpace(s)
	if s == "" {
		return nil, nil
	}
	if t, err := time.Parse(time.RFC3339, s); err == nil {
		return &t, nil
	}
	if t, err := time.Parse("2006-01-02", s); err == nil {
		return &t, nil
	}
	return nil, fmt.Errorf("%w: response_deadline must be RFC3339 or YYYY-MM-DD", ErrInvalidInput)
}

func parseUUIDPtr(s string) (*uuid.UUID, error) {
	s = strings.TrimSpace(s)
	if s == "" {
		return nil, nil
	}
	id, err := uuid.Parse(s)
	if err != nil {
		return nil, fmt.Errorf("%w: capture_owner_user_id must be a uuid", ErrInvalidInput)
	}
	return &id, nil
}