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

Source identity

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

Download complete source text ↓305 lines · 12.6 KB
package opportunities

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"path/filepath"
	"strings"

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

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

// Document intake, directive Section 6 screen 3, and the missing half of Phase 2 Gate 1.
//
// This is the path a customer's solicitation package actually enters the system by. Until now
// opportunity_documents and opportunity_document_chunks were SCHEMA WITHOUT CODE: correct tables carrying
// provenance bound at origin, and not one line of application code that wrote to either. The document trial
// rehearsal found that, and this closes it.
//
// THE RULES THIS FILE EXISTS TO ENFORCE, all of them refusals rather than conveniences:
//
//	the bytes go to object storage, never into a column
//	sha256 is computed here and is the duplicate key WITHIN an opportunity, so re uploading the same file
//	  is recognised rather than silently duplicated
//	the media type allowlist is closed, and an unknown type is refused with a reason a human can act on
//	a size cap is enforced BEFORE the bytes are stored, not after
//	provenance is NEVER passed by this code. Migration 0081 makes it born tenant_upload, and a document row
//	  that could declare its own provenance would defeat the entire egress boundary
//	the malware scan hook sits between the upload and extraction_status pending, which is the ordering the
//	  schema comment specifies
//
// Everything runs inside RunInTenantTx. Writing a tenant row on a bare pool against forced RLS is the
// specific bug that made PATCH /me/profile return 404 on the caller's own account, and it is not repeated.

// MaxDocumentBytes caps a single upload. Deliberately a constant rather than configuration: a cap that can
// be raised by an environment variable gets raised at 2am by whoever is annoyed, and the reason it exists
// (a solicitation package is documents, not disk images) does not vary by deployment.
const MaxDocumentBytes = 64 << 20 // 64 MiB

// allowedMediaTypes is CLOSED. An unknown type fails rather than being stored hopefully, for the same
// reason the egress class registry is closed: a permissive default is indistinguishable from no rule.
var allowedMediaTypes = map[string]string{
	"application/pdf": ".pdf",
	"application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx",
	"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet":       ".xlsx",
	"application/msword":       ".doc",
	"application/vnd.ms-excel": ".xls",
	"text/plain":               ".txt",
	"text/html":                ".html",
	"application/rtf":          ".rtf",
	"application/zip":          ".zip",
}

var validDocumentTypes = map[string]bool{
	"solicitation": true, "amendment": true, "sow": true, "pws": true, "soo": true,
	"labor_categories": true, "pricing_template": true, "evaluation_criteria": true,
	"security": true, "q_and_a": true, "past_performance": true, "subcontract_scope": true,
	"customer_scope": true, "attachment": true, "other": true,
}

var (
	// ErrDocumentTooLarge and friends are values rather than strings so a handler can map them to a status
	// code without matching on prose.
	ErrDocumentTooLarge    = errors.New("opportunities: document exceeds the size cap")
	ErrDocumentEmpty       = errors.New("opportunities: document is empty")
	ErrMediaTypeRefused    = errors.New("opportunities: media type is not in the allowlist")
	ErrDocumentTypeInvalid = errors.New("opportunities: document_type is not in the vocabulary")
	ErrDuplicateDocument   = errors.New("opportunities: an identical document is already attached to this opportunity")
	ErrOpportunityUnknown  = errors.New("opportunities: no such tracked opportunity for this organization")
)

// Document is the app facing view of one uploaded document.
type Document struct {
	ID               string `json:"id"`
	OpportunityID    string `json:"opportunity_id"`
	DocumentType     string `json:"document_type"`
	FileName         string `json:"file_name"`
	MediaType        string `json:"media_type"`
	SizeBytes        int64  `json:"size_bytes"`
	SHA256           string `json:"sha256"`
	SourceType       string `json:"source_type"`
	ExtractionStatus string `json:"extraction_status"`
	ExtractionError  string `json:"extraction_error_code"`
	IsCurrent        bool   `json:"is_current"`
	Provenance       string `json:"provenance"`
	CreatedAt        string `json:"created_at"`
}

// UploadInput carries one file. Provenance is deliberately ABSENT from this struct: there is no field for a
// caller to set, so the only way a document could claim to be public is to ship a migration.
type UploadInput struct {
	OpportunityID string
	DocumentType  string
	FileName      string
	MediaType     string
	Bytes         []byte
}

// MalwareScanner is the integration point the schema comment reserves. It runs BETWEEN the bytes arriving
// and the row reaching extraction_status pending, so an infected file never becomes work for the pipeline.
//
// The default is NoopScanner, and it is NAMED noop rather than quietly absent, because a scanner that does
// nothing while looking like a scanner is worse than an obvious hole.
type MalwareScanner interface {
	Scan(ctx context.Context, name string, data []byte) error
}

type NoopScanner struct{}

func (NoopScanner) Scan(context.Context, string, []byte) error { return nil }

// DocumentService owns the intake path.
type DocumentService struct {
	db      *tenant.DB
	store   storage.Storer
	scanner MalwareScanner
	audit   *audit.Writer
}

// created_at is cast to text because Document.CreatedAt is a string, and pgx refuses to scan a timestamptz
// into one. The scope service alongside this one already casts its timestamps and this list did not, so
// every upload failed 500 at the RETURNING clause after the bytes were already stored. No pure test could
// see it: the column list is only ever executed against a real database.
const documentCols = `id::text, opportunity_id::text, document_type, file_name, media_type,
	size_bytes, sha256, source_type, extraction_status, extraction_error_code, is_current, provenance,
	created_at::text`

func NewDocumentService(db *tenant.DB, store storage.Storer, scanner MalwareScanner, aw *audit.Writer) *DocumentService {
	if scanner == nil {
		scanner = NoopScanner{}
	}
	return &DocumentService{db: db, store: store, scanner: scanner, audit: aw}
}

// Upload validates, scans, stores the bytes, and records the row. The ORDER is the design:
// validate, then scan, then store, then record. A row that exists before its bytes do is a row the
// extraction pipeline will pick up and fail on, and a stored blob with no row is garbage nobody can find.
func (s *DocumentService) Upload(ctx context.Context, userID uuid.UUID, in UploadInput) (Document, error) {
	var out Document
	if !s.db.Ready() {
		return out, ErrNoDatabase
	}
	// THE ORG COMES FROM THE VALIDATED CONTEXT, never from a parameter. Passing it in would make the caller
	// the authority on tenancy, which is the whole thing RLS exists to stop.
	orgStr, ok := tenant.OrgFromContext(ctx)
	if !ok {
		return out, ErrNoDatabase
	}
	orgID, err := uuid.Parse(orgStr)
	if err != nil {
		return out, ErrNoDatabase
	}

	if len(in.Bytes) == 0 {
		return out, ErrDocumentEmpty
	}
	if len(in.Bytes) > MaxDocumentBytes {
		return out, fmt.Errorf("%w: %d bytes, cap is %d", ErrDocumentTooLarge, len(in.Bytes), MaxDocumentBytes)
	}
	mt := normalizeMediaType(in.MediaType, in.FileName)
	if _, ok := allowedMediaTypes[mt]; !ok {
		return out, fmt.Errorf("%w: %q", ErrMediaTypeRefused, mt)
	}
	docType := strings.TrimSpace(in.DocumentType)
	if docType == "" {
		docType = "other"
	}
	if !validDocumentTypes[docType] {
		return out, fmt.Errorf("%w: %q", ErrDocumentTypeInvalid, docType)
	}
	oppID, perr := uuid.Parse(strings.TrimSpace(in.OpportunityID))
	if perr != nil {
		return out, ErrOpportunityUnknown
	}

	// THE MALWARE HOOK, before anything is persisted anywhere.
	if err := s.scanner.Scan(ctx, in.FileName, in.Bytes); err != nil {
		return out, fmt.Errorf("opportunities: malware scan refused %q: %w", in.FileName, err)
	}

	sum := storage.SHA256Hex(in.Bytes)
	key := fmt.Sprintf("org/%s/opportunity/%s/%s%s", orgID, oppID, sum, allowedMediaTypes[mt])

	err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		// The opportunity must exist FOR THIS ORG. Checked inside the tenant transaction so RLS answers the
		// question rather than the application believing it.
		var exists bool
		if e := tx.QueryRow(ctx,
			`SELECT EXISTS (SELECT 1 FROM tracked_opportunities WHERE id=$1)`,
			oppID).Scan(&exists); e != nil {
			return e
		}
		if !exists {
			return ErrOpportunityUnknown
		}

		// DUPLICATE DETECTION BY SHA256, scoped to the opportunity. The same attachment legitimately appears
		// under two different opportunities, so a global unique would be wrong.
		var dupID string
		e := tx.QueryRow(ctx,
			`SELECT id::text FROM opportunity_documents
			  WHERE opportunity_id=$1 AND sha256=$2 AND is_current
			  LIMIT 1`, oppID, sum).Scan(&dupID)
		if e == nil {
			return fmt.Errorf("%w: %s", ErrDuplicateDocument, dupID)
		}
		if !errors.Is(e, pgx.ErrNoRows) {
			return e
		}

		// The bytes go to storage BEFORE the row, so a row never points at nothing. If the insert then
		// fails, the orphan blob is addressed by its own content hash and is re used by the next attempt
		// rather than leaking a new object each time.
		if pe := s.store.Put(ctx, key, in.Bytes); pe != nil {
			return fmt.Errorf("opportunities: storing document bytes: %w", pe)
		}

		// PROVENANCE IS NOT IN THIS INSERT. Migration 0081 stamps tenant_upload on the row regardless of
		// what anyone supplies, and omitting the column here makes that visible at the call site.
		return tx.QueryRow(ctx, `
			INSERT INTO opportunity_documents
			  (organization_id, opportunity_id, document_type, file_name, media_type, storage_key,
			   size_bytes, sha256, source_type, uploaded_by_user_id, extraction_status)
			VALUES ($1,$2,$3,$4,$5,$6,$7,$8,'upload',$9,'pending')
			RETURNING `+documentCols,
			orgID, oppID, docType, filepath.Base(in.FileName), mt, key,
			int64(len(in.Bytes)), sum, userID,
		).Scan(&out.ID, &out.OpportunityID, &out.DocumentType, &out.FileName, &out.MediaType,
			&out.SizeBytes, &out.SHA256, &out.SourceType, &out.ExtractionStatus,
			&out.ExtractionError, &out.IsCurrent, &out.Provenance, &out.CreatedAt)
	})
	if err != nil {
		return Document{}, err
	}

	if s.audit != nil {
		nv, _ := json.Marshal(map[string]any{
			"file_name": out.FileName, "sha256": out.SHA256, "size_bytes": out.SizeBytes,
			"document_type": out.DocumentType, "provenance": out.Provenance,
		})
		_ = s.audit.System(ctx, audit.Event{
			Action: "opportunity_document.upload", ActorUserID: &userID,
			EntityType: "opportunity_document", EntityID: out.ID, NewValues: nv,
		})
	}
	return out, nil
}

// List returns the documents attached to one opportunity, current first.
func (s *DocumentService) List(ctx context.Context, opportunityID string) ([]Document, error) {
	if !s.db.Ready() {
		return nil, ErrNoDatabase
	}
	oppID, err := uuid.Parse(strings.TrimSpace(opportunityID))
	if err != nil {
		return nil, ErrOpportunityUnknown
	}
	out := make([]Document, 0)
	err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		// No organization_id predicate: RLS supplies it from the transaction's org GUC. Adding one here
		// would be a second, weaker copy of the rule that could drift out of agreement with the policy.
		rows, e := tx.Query(ctx, `SELECT `+documentCols+`
			  FROM opportunity_documents WHERE opportunity_id=$1
			  ORDER BY is_current DESC, created_at DESC`, oppID)
		if e != nil {
			return e
		}
		defer rows.Close()
		for rows.Next() {
			var d Document
			if se := rows.Scan(&d.ID, &d.OpportunityID, &d.DocumentType, &d.FileName, &d.MediaType,
				&d.SizeBytes, &d.SHA256, &d.SourceType, &d.ExtractionStatus,
				&d.ExtractionError, &d.IsCurrent, &d.Provenance, &d.CreatedAt); se != nil {
				return se
			}
			out = append(out, d)
		}
		return rows.Err()
	})
	return out, err
}

// normalizeMediaType prefers the declared type and falls back to the file extension, because browsers send
// application/octet-stream for a .docx more often than anyone would like. It never GUESSES from content: a
// sniffed type is a claim about bytes, and the allowlist is a decision about formats.
func normalizeMediaType(declared, fileName string) string {
	d := strings.ToLower(strings.TrimSpace(strings.SplitN(declared, ";", 2)[0]))
	if _, ok := allowedMediaTypes[d]; ok {
		return d
	}
	ext := strings.ToLower(filepath.Ext(fileName))
	for mt, e := range allowedMediaTypes {
		if e == ext {
			return mt
		}
	}
	return d
}