// Command corpus-harvest builds missionMATCH's own knowledge of federal opportunities from the documents
// attached to them. It has two roles and the same binary runs both.
//
// WHY TWO ROLES RATHER THAN ONE LOOP. The two halves of this job have opposite constraints.
//
//	FETCH is not where the computing happens. It is a tap paced by politeness: one request per second to
//	sam.gov, and every device in this federation egresses through a single public address, so running it
//	on three machines would present as three requests per second from one IP and earn a block rather than
//	a speedup. Exactly one fetcher runs. It is not a bottleneck worth parallelising because it is not
//	doing any work, it is waiting.
//
//	EMBED is where the computing happens, and it is fully liquid. Any number of workers on any number of
//	machines claim chunks with FOR UPDATE SKIP LOCKED, which Postgres makes atomic without an orchestrator,
//	a scheduler or a heartbeat. A node under load claims fewer rows; an idle node claims more. That is self
//	balancing by construction rather than by policy, and a node that dies mid claim simply loses its
//	transaction and the rows return to the queue.
//
// THE BINARY IS NEVER RETAINED. Measured against sam.gov, a notice carries 2.5 files and 2.3 MB on average,
// so the active federal landscape is roughly 69 GB of PDF. Keeping it would buy nothing: SAM is a permanent
// public store with stable urls, so the durable copy already exists. The fetcher downloads to memory,
// extracts, writes the prose and its citations, and drops the bytes. sha256 and source_url are recorded on
// every row so any original refetches and can be proved identical to what was learned from.
//
// EACH NODE EMBEDS AGAINST ITS OWN LOCAL MODEL. M4 and VALKYRIE are Apple Silicon and run the model on
// Metal; megamind-agi has no GPU and contributes at CPU speed. Pointing every worker at one endpoint would
// recreate the bottleneck the federation exists to remove, so the endpoint is per node and defaults to
// localhost.
//
// Env: MM_MIGRATOR_DATABASE_URL (writes the shared corpus), MM_EMBED_ENDPOINT (defaults to the local
// ollama), MM_EMBED_MODEL, MM_EMBED_DIMENSION.
//
// Build and run FIPS validated: CGO_ENABLED=0 GOFIPS140=v1.0.0 go build.
package main

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"strings"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/missionworkforce/missionmatch/internal/modules/opportunities"
	"github.com/missionworkforce/missionmatch/internal/platform/docextract"
	"github.com/missionworkforce/missionmatch/internal/platform/embeddings"
)

const (
	// resourceListURL and fileURL are the PUBLIC sam.gov endpoints. They take no API key, which is the
	// whole reason an attachment crawl is affordable: sam-sync spends the keyed quota on notices, and an
	// attachment crawl is exactly the workload that would exhaust a daily quota and take the notice feed
	// down with it.
	resourceListURL = "https://sam.gov/api/prod/opps/v3/opportunities/%s/resources"
	fileURL         = "https://sam.gov/api/prod/opps/v3/opportunities/resources/files/%s/download"

	// maxFileBytes refuses a file rather than pulling a 400 MB drawing set into memory to find it has no
	// prose in it. Recorded as too_large so it is a decision on the record, not a silent skip.
	maxFileBytes = 32 << 20

	// embedBatch is how many chunks one worker claims at a time. Small on purpose: a big claim held by a
	// slow node is work the fast nodes cannot take, which is the opposite of liquid.
	embedBatch = 32

	// embedRetries is how many consecutive failures a worker tolerates before it gives up and lets the
	// other nodes carry the queue. Five is enough to ride out a link that blinks and few enough that a
	// genuinely unreachable database still ends the run rather than looping forever.
	embedRetries = 5
)

func main() {
	role := flag.String("role", "embed", "fetch (one node only, paced) or embed (any number of nodes)")
	limit := flag.Int("limit", 0, "stop after this many notices (fetch) or chunks (embed), 0 means until dry")
	openOnly := flag.Bool("open-only", true, "fetch only notices whose response deadline has not passed")
	rate := flag.Duration("rate", time.Second, "minimum interval between sam.gov requests")
	dryRun := flag.Bool("dry-run", false, "fetch: list and size the work, download nothing")
	flag.Parse()

	dsn := firstNonEmpty(os.Getenv("MM_MIGRATOR_DATABASE_URL"), os.Getenv("MM_DATABASE_URL"))
	if dsn == "" {
		log.Fatal("corpus-harvest: MM_MIGRATOR_DATABASE_URL is required")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 12*time.Hour)
	defer cancel()

	// THE POOL IS CAPPED AT TWO BECAUSE THAT IS WHAT THIS WORKER NEEDS, and for no other reason.
	//
	// pgxpool defaults to the greater of four and the CPU count, so an eight core box opens eight
	// connections to claim one batch at a time inside one transaction. Two is honest sizing: one working
	// connection, and one so a retry does not wait for it to be returned.
	//
	// It is recorded plainly that capping the pool was once believed to be the fix for the unreachable
	// wired path. It was not. A single detached run succeeded with the cap at one and causation was read
	// into one observation; the same run failed at two, and failed again with nothing else on the network.
	// The real cause was the interface scoped route, and it is addressed by the bound dialer below.
	cfg, err := pgxpool.ParseConfig(dsn)
	if err != nil {
		log.Fatalf("corpus-harvest: parse dsn: %v", err)
	}
	cfg.MaxConns = 2
	cfg.MinConns = 0

	// MM_BIND_INTERFACE names the wired interface this node should reach the federation on. Setting it
	// makes every socket this command opens carry IP_BOUND_IF, which is what lets it use the interface
	// scoped route macOS installs when a machine has both a wired default and a Tailscale tunnel. Leaving
	// it unset is the ordinary case and changes nothing.
	bindIface := strings.TrimSpace(os.Getenv("MM_BIND_INTERFACE"))
	dialer, err := bindToInterface(bindIface)
	if err != nil {
		log.Fatalf("corpus-harvest: %v", err)
	}
	if dialer != nil {
		cfg.ConnConfig.DialFunc = dialer
		log.Printf("corpus-harvest: sockets bound to %s", bindIface)
	}

	pool, err := pgxpool.NewWithConfig(ctx, cfg)
	if err != nil {
		log.Fatalf("corpus-harvest: pool: %v", err)
	}
	defer pool.Close()

	host, _ := os.Hostname()
	switch *role {
	case "fetch":
		if err := runFetch(ctx, pool, host, *limit, *openOnly, *rate, *dryRun); err != nil {
			log.Fatalf("corpus-harvest: fetch: %v", err)
		}
	case "embed":
		if err := runEmbed(ctx, pool, host, *limit); err != nil {
			log.Fatalf("corpus-harvest: embed: %v", err)
		}
	default:
		log.Fatalf("corpus-harvest: unknown role %q, want fetch or embed", *role)
	}
}

// ---------------------------------------------------------------------------------------------------
// FETCH: one node, paced, downloads nothing twice and keeps nothing.
// ---------------------------------------------------------------------------------------------------

// resource is one attachment as sam.gov actually describes it.
//
// THE FIELDS BEYOND THE IDENTIFIERS ARE GUARDS, NOT DECORATION. A first cut of this took the list at face
// value and produced four rows whose resource id was empty, because the payload nests: the
// opportunityAttachmentList is a list of GROUPS, each carrying its own attachments array. That failure was
// visible only by running it against sam.gov, and the empty download urls it wrote are the reason the
// parsing is now pinned to the real shape rather than an assumed one.
type resource struct {
	ResourceID       string `json:"resourceId"`
	Name             string `json:"name"`
	Type             string `json:"type"`
	MimeType         string `json:"mimeType"`
	Size             int64  `json:"size"`
	FileExists       string `json:"fileExists"`
	DeletedFlag      string `json:"deletedFlag"`
	AccessLevel      string `json:"accessLevel"`
	AccessStatus     string `json:"accessStatus"`
	ExportControlled string `json:"exportControlled"`
}

// fetchable decides whether this attachment may be downloaded at all.
//
// EXPORT CONTROLLED MATERIAL IS NEVER FETCHED, and that is the load bearing line in this function. sam.gov
// marks some attachments as export controlled or non public, and a cleared workforce product pulling ITAR
// or EAR restricted documents into a corpus would be a serious problem regardless of how the crawl was
// intended. The default is REFUSE: anything that does not positively say it is public and unrestricted is
// left alone.
func (r resource) fetchable() (bool, string) {
	if r.ResourceID == "" {
		return false, "no resource id"
	}
	if r.ExportControlled != "" && r.ExportControlled != "0" {
		return false, "export controlled, deliberately not fetched"
	}
	if r.AccessLevel != "" && !strings.EqualFold(r.AccessLevel, "public") {
		return false, "not a public attachment"
	}
	if r.AccessStatus != "" && !strings.EqualFold(r.AccessStatus, "public") {
		return false, "not publicly released"
	}
	if r.DeletedFlag == "1" {
		return false, "withdrawn by the issuing office"
	}
	if r.FileExists == "0" {
		return false, "the file is gone from sam.gov"
	}
	return true, ""
}

func runFetch(ctx context.Context, pool *pgxpool.Pool, host string, limit int, openOnly bool, rate time.Duration, dryRun bool) error {
	// Notices that have never been looked at. A notice with any corpus_documents row has been visited, so
	// a rerun continues rather than starting over, which matters on a crawl measured in days.
	q := `
		SELECT o.id::text, o.source_record_id, coalesce(o.title,'')
		  FROM opportunity_records o
		 WHERE o.source_record_id IS NOT NULL AND o.source_record_id <> ''
		   AND NOT EXISTS (SELECT 1 FROM corpus_documents d WHERE d.opportunity_record_id = o.id)`
	if openOnly {
		q += ` AND o.response_deadline >= now()`
	}
	q += ` ORDER BY o.response_deadline NULLS LAST`
	if limit > 0 {
		q += fmt.Sprintf(" LIMIT %d", limit)
	}

	rows, err := pool.Query(ctx, q)
	if err != nil {
		return err
	}
	type target struct{ id, notice, title string }
	var targets []target
	for rows.Next() {
		var t target
		if err := rows.Scan(&t.id, &t.notice, &t.title); err != nil {
			rows.Close()
			return err
		}
		targets = append(targets, t)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return err
	}

	log.Printf("corpus-harvest: fetch on %s, %d notice(s) unvisited, open_only=%t rate=%s dry_run=%t",
		host, len(targets), openOnly, rate, dryRun)
	if len(targets) == 0 {
		return nil
	}

	client := &http.Client{Timeout: 90 * time.Second}
	var files, chunks, skipped int
	var bytesIn int64
	tick := time.NewTicker(rate)
	defer tick.Stop()

	for i, t := range targets {
		<-tick.C
		res, err := listResources(ctx, client, t.notice)
		if err != nil {
			log.Printf("corpus-harvest: notice %s resource list failed: %v", t.notice, err)
			continue
		}
		if len(res) == 0 {
			// A notice with no attachments is a FACT worth recording, not an absence to retry forever.
			// One row with status skipped stops this notice costing a request on every future pass.
			if !dryRun {
				if err := recordNoAttachments(ctx, pool, t.id); err != nil {
					log.Printf("corpus-harvest: notice %s: %v", t.notice, err)
				}
			}
			continue
		}
		for _, r := range res {
			// The guard runs BEFORE the request, so a restricted attachment costs sam.gov nothing and
			// never touches this machine. The refusal is recorded so the corpus can account for what it
			// deliberately did not read.
			if ok, why := r.fetchable(); !ok {
				if !dryRun {
					if err := recordRefusal(ctx, pool, t.id, r, why); err != nil {
						log.Printf("corpus-harvest: notice %s refusal: %v", t.notice, err)
					}
				}
				skipped++
				continue
			}
			if dryRun {
				files++
				bytesIn += r.Size
				continue
			}
			<-tick.C
			n, c, err := fetchOne(ctx, pool, client, t.id, t.notice, r)
			if err != nil {
				log.Printf("corpus-harvest: notice %s file %s: %v", t.notice, r.ResourceID, err)
				continue
			}
			files++
			bytesIn += n
			chunks += c
		}
		if (i+1)%25 == 0 {
			log.Printf("corpus-harvest: %d/%d notices, %d files, %d chunks, %.1f MB fetched and discarded",
				i+1, len(targets), files, chunks, float64(bytesIn)/(1<<20))
		}
	}
	log.Printf("corpus-harvest: fetch DONE, %d notices, %d files, %d chunks, %d refused, %.1f MB fetched and discarded",
		len(targets), files, chunks, skipped, float64(bytesIn)/(1<<20))
	return nil
}

func listResources(ctx context.Context, c *http.Client, noticeID string) ([]resource, error) {
	req, _ := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf(resourceListURL, noticeID), nil)
	req.Header.Set("User-Agent", "missionMATCH corpus harvest")
	resp, err := c.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("resource list status %d", resp.StatusCode)
	}
	body, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
	if err != nil {
		return nil, err
	}
	// THE LIST IS NESTED TWO DEEP. opportunityAttachmentList is a list of GROUPS, one per opportunity, and
	// each group carries its own attachments array. Reading the outer list as the attachments themselves
	// decodes without error and yields rows with every field empty, which is exactly what the first live
	// run wrote: four documents whose download url read files//download. The flat shape is still accepted
	// because sam.gov has changed this envelope before, but the nested one is the shape it serves today.
	var envelope struct {
		Embedded struct {
			Groups []struct {
				OpportunityID string     `json:"opportunityId"`
				Attachments   []resource `json:"attachments"`
			} `json:"opportunityAttachmentList"`
		} `json:"_embedded"`
		Resources []resource `json:"resources"`
	}
	if err := json.Unmarshal(body, &envelope); err != nil {
		return nil, fmt.Errorf("resource list is not the expected json: %w", err)
	}
	if len(envelope.Resources) > 0 {
		return envelope.Resources, nil
	}
	var out []resource
	for _, g := range envelope.Embedded.Groups {
		out = append(out, g.Attachments...)
	}
	return out, nil
}

// fetchOne downloads, extracts, writes and DISCARDS. The bytes exist only inside this function.
func fetchOne(ctx context.Context, pool *pgxpool.Pool, c *http.Client, recordID, noticeID string, r resource) (int64, int, error) {
	status, errText := "ok", ""
	var data []byte
	var sum string

	req, _ := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf(fileURL, r.ResourceID), nil)
	req.Header.Set("User-Agent", "missionMATCH corpus harvest")
	resp, err := c.Do(req)
	if err != nil {
		status, errText = "fetch_failed", err.Error()
	} else {
		defer resp.Body.Close()
		if resp.StatusCode != http.StatusOK {
			status, errText = "fetch_failed", fmt.Sprintf("status %d", resp.StatusCode)
		} else {
			data, err = io.ReadAll(io.LimitReader(resp.Body, maxFileBytes+1))
			switch {
			case err != nil:
				status, errText = "fetch_failed", err.Error()
			case int64(len(data)) > maxFileBytes:
				status, errText, data = "too_large", "over the per file cap", nil
			}
		}
	}

	var chunks []opportunities.Chunk
	if status == "ok" {
		h := sha256.Sum256(data)
		sum = hex.EncodeToString(h[:])
		text, xerr := docextract.ExtractText(ctx, r.Name, data)
		switch {
		case xerr != nil:
			status, errText = "parse_error", xerr.Error()
		case strings.TrimSpace(text) == "":
			status, errText = "no_prose", "nothing readable in this file"
		default:
			chunks = opportunities.ExtractDeterministic(text)
			if len(chunks) == 0 {
				status, errText = "no_prose", "no citable span found"
			}
		}
	}

	n := int64(len(data))
	written, err := writeDocument(ctx, pool, recordID, noticeID, r, sum, status, errText, n, chunks)
	// data goes out of scope here and is never written to disk. That is the compression.
	return n, written, err
}

func writeDocument(ctx context.Context, pool *pgxpool.Pool, recordID, noticeID string, r resource,
	sum, status, errText string, size int64, chunks []opportunities.Chunk) (int, error) {

	tx, err := pool.Begin(ctx)
	if err != nil {
		return 0, err
	}
	defer tx.Rollback(ctx)

	// sha256 and source_url are NOT NULL because a document whose original cannot be recovered is a claim
	// without evidence. A failed fetch still records the url and an empty hash it can be retried from.
	var docID string
	err = tx.QueryRow(ctx, `
		INSERT INTO corpus_documents
		    (opportunity_record_id, resource_id, file_name, media_type, size_bytes,
		     sha256, source_url, extract_status, extract_error, fetched_bytes, fetched_at)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, now())
		ON CONFLICT (opportunity_record_id, resource_id) DO UPDATE
		    SET extract_status = EXCLUDED.extract_status,
		        extract_error  = EXCLUDED.extract_error,
		        sha256         = EXCLUDED.sha256,
		        fetched_bytes  = EXCLUDED.fetched_bytes,
		        fetched_at     = now(),
		        updated_at     = now()
		RETURNING id::text`,
		recordID, r.ResourceID, r.Name, r.Type, nullableInt(r.Size),
		sum, fmt.Sprintf(fileURL, r.ResourceID), status, truncate(errText, 500), size).Scan(&docID)
	if err != nil {
		return 0, err
	}

	written := 0
	for _, c := range chunks {
		// embedding is left NULL on purpose. It is the queue the embed workers drain, and leaving it null
		// is what makes the expensive half of this job liquid across the federation.
		if _, err := tx.Exec(ctx, `
			INSERT INTO corpus_chunks
			    (document_id, page_number, section_label, text_content, content_hash,
			     extraction_method, confidence)
			VALUES ($1, $2, $3, $4, $5, $6, $7)`,
			docID, nullableInt(int64(c.PageNumber)), c.SectionLabel, c.Text, c.ContentHash,
			c.Method, c.Confidence); err != nil {
			return written, err
		}
		written++
	}
	return written, tx.Commit(ctx)
}

func recordNoAttachments(ctx context.Context, pool *pgxpool.Pool, recordID string) error {
	_, err := pool.Exec(ctx, `
		INSERT INTO corpus_documents
		    (opportunity_record_id, resource_id, file_name, sha256, source_url,
		     extract_status, extract_error, fetched_at)
		VALUES ($1, 'none', '', '', 'none', 'skipped', 'this notice has no attachments', now())
		ON CONFLICT (opportunity_record_id, resource_id) DO NOTHING`, recordID)
	return err
}

// ---------------------------------------------------------------------------------------------------
// EMBED: any number of nodes, self balancing, no orchestrator.
// ---------------------------------------------------------------------------------------------------

func runEmbed(ctx context.Context, pool *pgxpool.Pool, host string, limit int) error {
	// THE OPENAI COMPATIBLE PATH, not ollama's native one. Both are served on the same port and they take
	// different shapes: /api/embeddings takes a single prompt and returns one embedding, while
	// /v1/embeddings takes an input array and returns a data array, which is what the platform embedder
	// already speaks. Pointing at the native path decoded cleanly and returned zero vectors for a batch of
	// thirty two, which is a silent shape mismatch rather than an error, and both nodes failed identically
	// on it within a second of each other.
	endpoint := firstNonEmpty(os.Getenv("MM_EMBED_ENDPOINT"), "http://127.0.0.1:11434/v1/embeddings")
	model := firstNonEmpty(os.Getenv("MM_EMBED_MODEL"), "qllama/bge-small-en-v1.5")
	dim := 384
	// The HTTP client gets the SAME binding. Today each node embeds against its own ollama, but the
	// endpoint is configurable and pointing one node at another is exactly the liquid arrangement this
	// exists for, so the transport must be able to reach across the switch too.
	httpClient := &http.Client{Timeout: 60 * time.Second}
	if dialer, derr := bindToInterface(strings.TrimSpace(os.Getenv("MM_BIND_INTERFACE"))); derr == nil && dialer != nil {
		httpClient.Transport = &http.Transport{DialContext: dialer}
	}
	emb := embeddings.NewHTTP(endpoint, model, "", dim, httpClient)

	log.Printf("corpus-harvest: embed on %s against %s model=%s dim=%d", host, endpoint, model, dim)

	total := 0
	// A WORKER THAT DIES ON ONE TRANSIENT CONNECT IS NOT A FEDERATION MEMBER, it is a single point of
	// failure wearing a distributed hat. VALKYRIE reaches the database across a switch through a USB
	// ethernet adapter, and a detached run occasionally got EHOSTUNREACH on its first connect while ten
	// sequential runs in a held session succeeded every time. The adapter itself is clean, 51 million
	// packets with zero errors, so the loss is somewhere above it and is not worth pretending will never
	// happen. The queue is already safe against a worker vanishing mid batch, so the only thing missing was
	// the patience to try again.
	consecutiveFailures := 0
	for {
		n, err := embedOnce(ctx, pool, emb)
		if err != nil {
			consecutiveFailures++
			if consecutiveFailures > embedRetries {
				return fmt.Errorf("gave up after %d consecutive failures: %w", consecutiveFailures, err)
			}
			// Linear backoff, deliberately short. This is a transient network fault on a local switch, not
			// a rate limited API, so seconds are the right unit and there is nothing to be polite to.
			wait := time.Duration(consecutiveFailures) * 2 * time.Second
			log.Printf("corpus-harvest: %s attempt %d failed (%v), retrying in %s",
				host, consecutiveFailures, err, wait)
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(wait):
			}
			continue
		}
		consecutiveFailures = 0
		if n == 0 {
			log.Printf("corpus-harvest: embed DONE on %s, queue dry, %d chunk(s) embedded here", host, total)
			return nil
		}
		total += n
		if total%(embedBatch*8) == 0 {
			log.Printf("corpus-harvest: %s has embedded %d chunk(s)", host, total)
		}
		if limit > 0 && total >= limit {
			log.Printf("corpus-harvest: embed stopping at the requested limit, %d chunk(s) on %s", total, host)
			return nil
		}
	}
}

// embedOnce claims a batch and embeds it inside ONE transaction.
//
// FOR UPDATE SKIP LOCKED is the whole distribution mechanism. Postgres hands each worker a disjoint set
// with no coordination between them, so a second and third node need no configuration to join and no
// permission to leave. Holding the transaction open across the embedding calls is deliberate: if this node
// dies mid batch the transaction dies with it and the rows are immediately claimable by somebody else,
// which is the failure behaviour a lease with a timeout only approximates.
func embedOnce(ctx context.Context, pool *pgxpool.Pool, emb embeddings.Embedder) (int, error) {
	tx, err := pool.Begin(ctx)
	if err != nil {
		return 0, err
	}
	defer tx.Rollback(ctx)

	rows, err := tx.Query(ctx, `
		SELECT id, text_content FROM corpus_chunks
		 WHERE embedding IS NULL
		 ORDER BY id
		 FOR UPDATE SKIP LOCKED
		 LIMIT $1`, embedBatch)
	if err != nil {
		return 0, err
	}
	var ids []int64
	var texts []string
	for rows.Next() {
		var id int64
		var t string
		if err := rows.Scan(&id, &t); err != nil {
			rows.Close()
			return 0, err
		}
		ids = append(ids, id)
		texts = append(texts, t)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return 0, err
	}
	if len(ids) == 0 {
		return 0, nil
	}

	vecs, err := emb.EmbedBatch(ctx, texts)
	if err != nil {
		return 0, fmt.Errorf("embed batch of %d: %w", len(texts), err)
	}
	if len(vecs) != len(ids) {
		return 0, fmt.Errorf("embedder returned %d vectors for %d chunks", len(vecs), len(ids))
	}

	for i, id := range ids {
		if _, err := tx.Exec(ctx,
			`UPDATE corpus_chunks SET embedding = $2::halfvec WHERE id = $1`, id, vectorLiteral(vecs[i])); err != nil {
			return 0, err
		}
	}
	return len(ids), tx.Commit(ctx)
}

// vectorLiteral renders a vector in the text form pgvector parses for both vector and halfvec, so the
// worker needs no pgvector Go type registration to write a halfvec column.
func vectorLiteral(v []float32) string {
	var b strings.Builder
	b.WriteByte('[')
	for i, f := range v {
		if i > 0 {
			b.WriteByte(',')
		}
		fmt.Fprintf(&b, "%g", f)
	}
	b.WriteByte(']')
	return b.String()
}

func nullableInt(n int64) any {
	if n <= 0 {
		return nil
	}
	return n
}

func truncate(s string, n int) string {
	if len(s) <= n {
		return s
	}
	return s[:n]
}

func firstNonEmpty(vals ...string) string {
	for _, v := range vals {
		if s := strings.TrimSpace(v); s != "" {
			return s
		}
	}
	return ""
}

// recordRefusal writes the fact that an attachment was deliberately NOT read, and why.
//
// A corpus that silently omits what it refused cannot be audited. An export controlled document that was
// left alone is a decision the product made, and it belongs on the record next to the ones it did read.
func recordRefusal(ctx context.Context, pool *pgxpool.Pool, recordID string, r resource, why string) error {
	id := r.ResourceID
	if id == "" {
		id = "unknown-" + r.Name
	}
	_, err := pool.Exec(ctx, `
		INSERT INTO corpus_documents
		    (opportunity_record_id, resource_id, file_name, media_type, sha256, source_url,
		     extract_status, extract_error, fetched_at)
		VALUES ($1, $2, $3, $4, '', $5, 'skipped', $6, now())
		ON CONFLICT (opportunity_record_id, resource_id) DO UPDATE
		    SET extract_status = 'skipped', extract_error = EXCLUDED.extract_error, updated_at = now()`,
		recordID, id, r.Name, r.MimeType, fmt.Sprintf(fileURL, r.ResourceID), truncate(why, 500))
	return err
}