Files
bwilliams b0230e6847 Replaced the Title filter with a general search, run by the backend
GET /api/opportunities/search?q= returns the UUIDs of the matching
opportunities, and the frontend filters its list by them. Words without
a prompt search the title and description; title:, buyer:, buyer_id:
and cpv: search only their own field, and a leading ! leaves out what a
term matches. Repeats of one prompt are OR, everything else AND. CPV
terms are checked per code: cpv:72 !cpv:7221 needs a code starting 72
that doesn't start 7221.

Unknown prompts are reported, and every response carries the help that
the search box's "?" panel shows. Migration 00011 adds the description
and buyer name to the full-text index, and indexes CPV descriptions.
2026-10-22 13:10:25 +01:00

1085 lines
32 KiB
Go

// Package store persists opportunities and their OCDS releases in SQLite using
// sqlx and hand-written SQL. The SQL is in queries.go; this file holds the
// logic. Requires building with -tags sqlite_fts5.
package store
import (
"context"
"crypto/rand"
"crypto/sha256"
"database/sql"
"encoding/hex"
"errors"
"fmt"
"strings"
"time"
"unicode"
"github.com/jmoiron/sqlx"
"github.com/mattn/go-sqlite3"
"git.avsdev.uk/Go/job-search/internal/types"
)
// Errors returned by the store.
var (
// ErrNotFound is returned when a requested opportunity does not exist.
ErrNotFound = errors.New("store: not found")
// ErrUnknownBatch is returned by ListChangedOpportunities for a hash
// that names no change batch.
ErrUnknownBatch = errors.New("store: unknown change batch")
// ErrNeedsMigration is returned by Open when the database has pending
// migrations. Apply them with the migrate command, or pass
// WithAutoMigrate.
ErrNeedsMigration = errors.New("store: database needs migrating")
// ErrSchemaTooNew is returned by Open when the database has migrations
// this build doesn't know about, i.e. it was migrated by a newer build.
ErrSchemaTooNew = errors.New(
"store: database schema is newer than this build",
)
// ErrWouldEmptyDatabase is returned by StepDown when the migration would
// drop all of the tables and thus empty the database.
ErrWouldEmptyDatabase = errors.New(
"store: migration cancelled; database would be emptied by undoing all migrations",
)
)
// DefaultListLimit is used when OpportunityFilter.Limit is zero or negative.
const DefaultListLimit = 50
// Store is the opportunity database. Open it with Open; it is safe for
// concurrent use.
type Store struct {
db *sqlx.DB
// changeBatchRetention is how long change batches are kept; 0 or less
// means for ever (see WithChangeBatchRetention).
changeBatchRetention time.Duration
}
// SaveMultipleResult reports what SaveOpportunities changed.
type SaveMultipleResult struct {
// NewOpportunities is the number of opportunities that were not already
// stored.
NewOpportunities int
// NewReleases is the number of releases that were not already stored.
NewReleases int
// UpdatedOpportunities is incremented for each opportunity row and its
// classifications that are updated (when every release in the save was older
// than the one the stored opportunity was last updated from). It does not
// include new opportunities.
UpdatedOpportunities int
// OpportunityResults is a slice of the results of each opportunity saved.
OpportunityResults []SaveResult
// AddedClassifications are the classification codes that no stored
// opportunity had before the save and at least one has after it, ordered
// by code; nil if there are none.
AddedClassifications []types.ClassificationCode
// RemovedClassificationIDs are the ids of the classification codes that
// a stored opportunity had before the save and none has after it, in
// code order; nil if there are none.
RemovedClassificationIDs []int64
// ChangeBatch is the hash of the change batch recording the
// opportunities the save stored for the first time or gave new releases
// (see ListChangedOpportunities); "" if there are none.
ChangeBatch string
}
// SaveResult reports what SaveOpportunity changed.
type SaveResult struct {
// NewOpportunity is true if the opportunity was not already stored.
NewOpportunity bool
// OpportunityUpdated is true if the opportunity row and its classifications
// were written. It is false when every release in the save was older than the
// one the stored opportunity was last updated from.
OpportunityUpdated bool
// NewReleases is the number of releases that were not already stored.
NewReleases int
}
// MarkFilter filters ListOpportunities on one of an opportunity's marks.
type MarkFilter int
// Values of MarkFilter.
const (
// MarkAny does not filter. It is the zero value.
MarkAny MarkFilter = iota
// MarkSet matches opportunities with the mark set.
MarkSet
// MarkUnset matches opportunities with the mark unset, including those
// with no marks stored.
MarkUnset
)
// OpportunityFilter selects opportunities for ListOpportunities. Zero-valued
// fields do not filter.
type OpportunityFilter struct {
Status string
// ClassificationCode matches opportunities with a classification of this code
// in any scheme.
ClassificationCode string
// Title is free text matched against opportunity titles: every word must
// appear, and each word also matches as a prefix ("cloud migr" matches
// "Cloud migration services").
Title string
// ReleasedFrom (inclusive) and ReleasedBefore (exclusive) bound
// Opportunity.LatestReleaseDate.
ReleasedFrom time.Time
ReleasedBefore time.Time
// Each mark filter matches Opportunity.Marks' field of the same name.
Hidden MarkFilter
Skipped MarkFilter
Interested MarkFilter
BidSubmitted MarkFilter
Won MarkFilter
Lost MarkFilter
// Limit defaults to DefaultListLimit. Results are ordered newest
// LatestReleaseDate first.
Limit int
Offset int
}
func (f OpportunityFilter) limit() int {
if f.Limit <= 0 {
return DefaultListLimit
}
return f.Limit
}
// prepareOpportunity validates opportunity and normalises it in place before
// saving: child rows get the opportunity's UUID and their release id, times
// are converted to UTC (so stored timestamps compare correctly as text), and
// LatestReleaseDate becomes the newest release date in the save.
func prepareOpportunity(opportunity *types.Opportunity) error {
if opportunity == nil {
return errors.New("store: nil opportunity")
}
if opportunity.UUID == "" {
return errors.New("store: opportunity has no UUID")
}
opportunity.Marks.UUID = opportunity.UUID
if err := opportunity.Marks.Validate(); err != nil {
return fmt.Errorf("opportunity %s: %w", opportunity.UUID, err)
}
latest := opportunity.LatestReleaseDate
for i := range opportunity.Releases {
release := &opportunity.Releases[i]
if release.ReleaseID == "" {
return fmt.Errorf(
"store: opportunity %s: release %d has no release id",
opportunity.UUID,
i,
)
}
release.UUID = opportunity.UUID
release.ReleaseDate = release.ReleaseDate.UTC()
if release.ReleaseDate.After(latest) {
latest = release.ReleaseDate
}
for tagIndex := range release.Tags {
release.Tags[tagIndex].UUID = opportunity.UUID
release.Tags[tagIndex].ReleaseID = release.ReleaseID
}
}
if latest.IsZero() {
return fmt.Errorf(
"store: opportunity %s has no release date", opportunity.UUID,
)
}
opportunity.LatestReleaseDate = latest.UTC()
opportunity.StartDate = utcPtr(opportunity.StartDate)
opportunity.EndDate = utcPtr(opportunity.EndDate)
opportunity.ClosingDate = utcPtr(opportunity.ClosingDate)
return nil
}
func utcPtr(when *time.Time) *time.Time {
if when == nil {
return nil
}
utc := when.UTC()
return &utc
}
// ftsQuery turns free text into an FTS5 query that requires every word, each
// as a prefix match. Words are split the same way the unicode61 tokenizer
// splits them and each is quoted, so user input can never be interpreted as
// FTS5 syntax (operators, column filters, unbalanced quotes). It returns ""
// when the text contains no words.
func ftsQuery(text string) string {
words := strings.FieldsFunc(text, func(char rune) bool {
return !unicode.IsLetter(char) && !unicode.IsNumber(char)
})
for i, word := range words {
words[i] = `"` + word + `"*`
}
return strings.Join(words, " ")
}
// Open opens the database at path. The schema must be exactly at this
// build's latest migration: Open returns ErrNeedsMigration if migrations are
// pending (run the migrate command first), or ErrSchemaTooNew if the database
// was migrated by a newer build. WithAutoMigrate applies pending migrations
// instead.
//
// A database file that doesn't exist is created, and then needs migrating.
func Open(ctx context.Context, path string, opts ...Option) (*Store, error) {
var settings options
for _, option := range opts {
option(&settings)
}
sqldb, err := OpenDB(path)
if err != nil {
return nil, err
}
if err := sqldb.PingContext(ctx); err != nil {
return nil, errors.Join(err, sqldb.Close())
}
if err := prepareSchema(ctx, sqldb, settings); err != nil {
return nil, errors.Join(err, sqldb.Close())
}
return &Store{
db: sqlx.NewDb(sqldb, "sqlite3"),
changeBatchRetention: settings.changeBatchRetention,
}, nil
}
// Close closes the database.
func (s *Store) Close() error { return s.db.Close() }
// Reads select whole rows ("SELECT *", "t.*", "c.*"). sqlx rejects a result
// column with no matching field, so a column added by a migration makes the
// reads fail loudly ("missing destination name ...") until it is mapped,
// instead of being silently ignored.
//
// opportunityRow gives the opportunities table's internal id column a home, so
// it stays out of types.Opportunity. sqlx maps the embedded Opportunity's
// fields as if declared here. The other tables have no internal columns and
// read straight into internal/types.
type opportunityRow struct {
ID int64 `db:"id"`
types.Opportunity
}
// changeBatch is a row of opportunity_change_batches: a batch of changed
// opportunities, named to clients by its Hash. ID and CreatedAt are internal
// to the store.
type changeBatch struct {
ID int64 `db:"id"`
Hash string `db:"hash"`
CreatedAt time.Time `db:"created_at"`
}
// SaveOpportunity stores opportunity and its releases in one transaction.
//
// Releases (and their tags) are write-once: ones already stored are skipped.
// The opportunity row and its classifications are replaced only if the
// opportunity's newest release is at least as new as the release the stored
// opportunity came from, so a late-arriving older release never undoes a
// newer one.
//
// opportunity.Marks are stored only for a new opportunity; existing marks are
// kept (use SetMarks to change them).
//
// A new opportunity, or one given new releases, is recorded in a change
// batch of its own (see ListChangedOpportunities).
//
// opportunity is normalised in place (see prepareOpportunity), and when it is
// written its classifications' IDs are filled in.
func (s *Store) SaveOpportunity(
ctx context.Context, opportunity *types.Opportunity,
) (SaveResult, error) {
if err := prepareOpportunity(opportunity); err != nil {
return SaveResult{}, err
}
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return SaveResult{}, err
}
defer tx.Rollback() //nolint:errcheck // no-op after Commit
result, _, err := saveOpportunity(ctx, tx, opportunity)
if err != nil {
return SaveResult{}, err
}
_, err = s.recordChangeBatch(
ctx,
tx,
changedUUIDs([]*types.Opportunity{opportunity}, []SaveResult{result}),
)
if err != nil {
return SaveResult{}, err
}
if err := tx.Commit(); err != nil {
return SaveResult{}, err
}
return result, nil
}
// SaveOpportunities stores each of opportunities and its releases, all in one
// transaction, as SaveOpportunity does for one.
//
// Releases (and their tags) are write-once: ones already stored are skipped.
// An opportunity row and its classifications are replaced only if the
// opportunity's newest release is at least as new as the release the stored
// opportunity came from, so a late-arriving older release never undoes a
// newer one.
//
// Marks are stored only for a new opportunity; existing marks are kept (use
// SetMarks to change them).
//
// Each opportunity is normalised in place (see prepareOpportunity), and when
// it is written its classifications' IDs are filled in. The result reports
// the classification codes the save brought into or out of use (see
// ListClassifications), and the hash of the change batch recording the
// opportunities that were new or given new releases (see
// ListChangedOpportunities).
func (s *Store) SaveOpportunities(
ctx context.Context, opportunities []*types.Opportunity,
) (SaveMultipleResult, error) {
for _, opportunity := range opportunities {
if err := prepareOpportunity(opportunity); err != nil {
return SaveMultipleResult{}, err
}
}
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return SaveMultipleResult{}, err
}
defer tx.Rollback() //nolint:errcheck // no-op after Commit
before, err := classificationsInUse(ctx, tx)
if err != nil {
return SaveMultipleResult{}, err
}
total := SaveMultipleResult{
OpportunityResults: make([]SaveResult, len(opportunities)),
}
for i, opportunity := range opportunities {
result, isNew, err := saveOpportunity(ctx, tx, opportunity)
if err != nil {
return SaveMultipleResult{}, err
}
total.OpportunityResults[i] = result
total.NewReleases += result.NewReleases
switch {
case isNew:
total.NewOpportunities++
case result.OpportunityUpdated:
total.UpdatedOpportunities++
}
}
after, err := classificationsInUse(ctx, tx)
if err != nil {
return SaveMultipleResult{}, err
}
added, removed := classificationChanges(before, after)
total.AddedClassifications = added
total.RemovedClassificationIDs = removed
total.ChangeBatch, err = s.recordChangeBatch(
ctx, tx, changedUUIDs(opportunities, total.OpportunityResults),
)
if err != nil {
return SaveMultipleResult{}, err
}
if err := tx.Commit(); err != nil {
return SaveMultipleResult{}, err
}
return total, nil
}
// classificationsInUse returns the classification codes at least one
// opportunity has, ordered by code, as tx sees them.
func classificationsInUse(
ctx context.Context, tx *sqlx.Tx,
) ([]types.ClassificationCode, error) {
var classifications []types.ClassificationCode
err := tx.SelectContext(ctx, &classifications, selectClassificationsInUse)
if err != nil {
return nil, fmt.Errorf("listing classifications: %w", err)
}
return classifications, nil
}
// classificationChanges compares the classification codes in use before and
// after a save: it returns the codes in after but not before, and the ids of
// those in before but not after, each in the order they're given.
func classificationChanges(
before, after []types.ClassificationCode,
) ([]types.ClassificationCode, []int64) {
wasInUse := make(map[int64]bool, len(before))
for _, classification := range before {
wasInUse[classification.ID] = true
}
isInUse := make(map[int64]bool, len(after))
for _, classification := range after {
isInUse[classification.ID] = true
}
var added []types.ClassificationCode
for _, classification := range after {
if !wasInUse[classification.ID] {
added = append(added, classification)
}
}
var removed []int64
for _, classification := range before {
if !isInUse[classification.ID] {
removed = append(removed, classification.ID)
}
}
return added, removed
}
// changedUUIDs returns the UUIDs of the opportunities that results, from
// saving them, show were new or given new releases. An opportunity whose row
// was only rewritten, such as one fetched again from its newest stored
// release, hasn't changed.
func changedUUIDs(
opportunities []*types.Opportunity, results []SaveResult,
) []string {
var uuids []string
for i, result := range results {
if result.NewOpportunity || result.NewReleases > 0 {
uuids = append(uuids, opportunities[i].UUID)
}
}
return uuids
}
// recordChangeBatch records a change batch of the opportunities uuids inside
// tx and returns its hash, or "" without recording anything if uuids is
// empty. Batches older than the store's retention are deleted (see
// WithChangeBatchRetention).
//
// The batch's created_at is the current time, or just after the newest
// batch's if that isn't later (the clock went back, or two batches came
// within its resolution), so batches are ordered by created_at. tx holds the
// write lock from its start (see dsn), so no other batch can be recorded in
// between.
func (s *Store) recordChangeBatch(
ctx context.Context, tx *sqlx.Tx, uuids []string,
) (string, error) {
if len(uuids) == 0 {
return "", nil
}
var latest changeBatch
err := tx.GetContext(ctx, &latest, selectLatestChangeBatch)
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return "", fmt.Errorf("reading the latest change batch: %w", err)
}
createdAt := time.Now().UTC()
if !createdAt.After(latest.CreatedAt) {
createdAt = latest.CreatedAt.Add(time.Nanosecond)
}
hash := newBatchHash(createdAt)
var batchID int64
err = tx.GetContext(ctx, &batchID, insertChangeBatch, hash, createdAt)
if err != nil {
return "", fmt.Errorf("recording a change batch: %w", err)
}
for _, uuid := range uuids {
_, err = tx.ExecContext(ctx, insertChangeBatchOpportunity, batchID, uuid)
if err != nil {
return "", fmt.Errorf(
"recording %s in a change batch: %w", uuid, err,
)
}
}
if s.changeBatchRetention > 0 {
cutoff := createdAt.Add(-s.changeBatchRetention)
_, err = tx.ExecContext(ctx, deleteChangeBatchesBefore, cutoff)
if err != nil {
return "", fmt.Errorf("deleting old change batches: %w", err)
}
}
return hash, nil
}
// newBatchHash returns the hash of a change batch created at createdAt: the
// hex SHA-256 hash of random bytes and the time. The random bytes make it
// unguessable, so a client can only name a batch it was given.
func newBatchHash(createdAt time.Time) string {
random := make([]byte, 32)
rand.Read(random) // crypto/rand.Read never returns an error
hash := sha256.Sum256(
append(random, createdAt.Format(time.RFC3339Nano)...),
)
return hex.EncodeToString(hash[:])
}
// saveOpportunity does the work of SaveOpportunity and SaveOpportunities for
// one prepared opportunity inside tx. isNew reports that the opportunity was
// not already stored.
func saveOpportunity(
ctx context.Context, tx *sqlx.Tx, opportunity *types.Opportunity,
) (SaveResult, bool, error) {
var (
result SaveResult
stored time.Time
)
err := tx.GetContext(
ctx, &stored, selectOpportunityLatestReleaseDate, opportunity.UUID,
)
switch {
case errors.Is(err, sql.ErrNoRows):
result.NewOpportunity = true
result.OpportunityUpdated = true
case err != nil:
return SaveResult{}, false, fmt.Errorf(
"reading opportunity %s: %w", opportunity.UUID, err,
)
default:
result.OpportunityUpdated = !opportunity.LatestReleaseDate.Before(stored)
}
if result.OpportunityUpdated {
if err := writeOpportunity(ctx, tx, opportunity); err != nil {
return SaveResult{}, false, fmt.Errorf(
"writing opportunity %s: %w", opportunity.UUID, err,
)
}
}
for _, release := range opportunity.Releases {
inserted, err := insertReleaseWithTags(ctx, tx, release)
if err != nil {
return SaveResult{}, false, err
}
if inserted {
result.NewReleases++
}
}
return result, result.NewOpportunity, nil
}
// insertReleaseWithTags stores release and its tags unless it is already
// stored, and reports whether it was inserted.
func insertReleaseWithTags(
ctx context.Context, tx *sqlx.Tx, release types.OpportunityRelease,
) (bool, error) {
result, err := tx.NamedExecContext(ctx, insertRelease, release)
if err != nil {
return false, fmt.Errorf(
"inserting release %s/%s: %w",
release.UUID,
release.ReleaseID,
err,
)
}
if affected, err := result.RowsAffected(); err != nil || affected == 0 {
return false, err // affected == 0: already stored
}
for _, tag := range release.Tags {
if _, err := tx.NamedExecContext(ctx, insertReleaseTag, tag); err != nil {
return false, fmt.Errorf(
"inserting tag for %s/%s: %w",
release.UUID,
release.ReleaseID,
err,
)
}
}
return true, nil
}
// writeOpportunity upserts opportunity's row, gives it a marks row if it has
// none (existing marks are kept), and replaces its classifications, filling in
// their IDs.
func writeOpportunity(
ctx context.Context, tx *sqlx.Tx, opportunity *types.Opportunity,
) (err error) {
_, err = tx.NamedExecContext(ctx, upsertOpportunity, opportunity)
if err != nil {
return err
}
_, err = tx.NamedExecContext(ctx, insertOpportunityMarks, opportunity.Marks)
if err != nil {
return err
}
_, err = tx.ExecContext(
ctx, deleteOpportunityClassifications, opportunity.UUID,
)
if err != nil {
return err
}
statement, err := tx.PrepareNamedContext(
ctx, upsertClassificationCodeReturningID,
)
if err != nil {
return err
}
defer func() {
err = errors.Join(err, statement.Close())
}()
for i := range opportunity.Classifications {
classification := &opportunity.Classifications[i]
err = statement.GetContext(ctx, &classification.ID, classification)
if err != nil {
return err
}
_, err = tx.ExecContext(
ctx, insertOpportunityClassification, opportunity.UUID, classification.ID,
)
if err != nil {
return err
}
}
return nil
}
// SetMarks replaces the marks of the opportunity marks.UUID, or returns
// ErrNotFound. Marks that fail marks.Validate (e.g. types.ErrWonAndLost) are
// not stored. The opportunity is recorded in a change batch of its own (see
// ListChangedOpportunities), so a client catching up sees its new marks.
func (s *Store) SetMarks(
ctx context.Context, marks types.OpportunityMarks,
) error {
if marks.UUID == "" {
return errors.New("store: marks have no UUID")
}
if err := marks.Validate(); err != nil {
return fmt.Errorf("opportunity %s: %w", marks.UUID, err)
}
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback() //nolint:errcheck // no-op after Commit
_, err = tx.NamedExecContext(ctx, upsertOpportunityMarks, marks)
if isForeignKeyError(err) {
return ErrNotFound
}
if err != nil {
return err
}
_, err = s.recordChangeBatch(ctx, tx, []string{marks.UUID})
if err != nil {
return err
}
return tx.Commit()
}
// isForeignKeyError reports whether err is SQLite rejecting a write that
// breaks a foreign key, e.g. one referring to a missing opportunity.
func isForeignKeyError(err error) bool {
var sqliteErr sqlite3.Error
return errors.As(err, &sqliteErr) &&
sqliteErr.ExtendedCode == sqlite3.ErrConstraintForeignKey
}
// GetOpportunity returns the opportunity with its classifications, marks and
// all of its releases (with tags), or ErrNotFound.
func (s *Store) GetOpportunity(
ctx context.Context, uuid string,
) (*types.Opportunity, error) {
var row opportunityRow
err := s.db.GetContext(ctx, &row, selectOpportunityByUUID, uuid)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrNotFound
}
if err != nil {
return nil, err
}
opportunities := []types.Opportunity{row.Opportunity}
if err := s.loadClassifications(ctx, opportunities); err != nil {
return nil, err
}
if err := s.loadMarks(ctx, opportunities); err != nil {
return nil, err
}
if opportunities[0].Releases, err = s.ListReleases(ctx, uuid); err != nil {
return nil, err
}
return &opportunities[0], nil
}
// ListOpportunities returns opportunities matching filter, with their
// classifications and marks but without releases (use ListReleases or
// GetOpportunity for those).
func (s *Store) ListOpportunities(
ctx context.Context, filter OpportunityFilter,
) ([]types.Opportunity, error) {
query, args := listOpportunitiesQuery(filter)
var rows []opportunityRow
if err := s.db.SelectContext(ctx, &rows, query, args...); err != nil {
return nil, err
}
return s.opportunitiesFromRows(ctx, rows)
}
// LatestChangeBatch returns the hash of the newest change batch, or "" if
// none has been recorded. Read before listing opportunities, it is the batch
// to give ListChangedOpportunities later to get what changed since.
func (s *Store) LatestChangeBatch(ctx context.Context) (string, error) {
var latest changeBatch
err := s.db.GetContext(ctx, &latest, selectLatestChangeBatch)
if errors.Is(err, sql.ErrNoRows) {
return "", nil
}
if err != nil {
return "", fmt.Errorf("reading the latest change batch: %w", err)
}
return latest.Hash, nil
}
// ListChangedOpportunities returns the opportunities recorded in the change
// batches created after the batch with the hash after, once each, as
// ListOpportunities returns them but unpaged, and the hash of the newest of
// those batches, to give it next time. With no newer batches, it returns no
// opportunities and after. An unknown hash is ErrUnknownBatch.
//
// Unless releasedFrom is zero, only opportunities whose latest release is on
// or after it are returned (as OpportunityFilter.ReleasedFrom); the hash
// returned is the same either way.
//
// Batches recorded while it runs are left for next time, so none is missed:
// only those up to the newest batch when it starts are read.
func (s *Store) ListChangedOpportunities(
ctx context.Context, after string, releasedFrom time.Time,
) ([]types.Opportunity, string, error) {
var since changeBatch
err := s.db.GetContext(ctx, &since, selectChangeBatchByHash, after)
if errors.Is(err, sql.ErrNoRows) {
return nil, "", ErrUnknownBatch
}
if err != nil {
return nil, "", fmt.Errorf("reading change batch %s: %w", after, err)
}
var latest changeBatch
err = s.db.GetContext(ctx, &latest, selectLatestChangeBatch)
if err != nil {
return nil, "", fmt.Errorf("reading the latest change batch: %w", err)
}
query := selectOpportunitiesChangedBetween
args := []any{since.CreatedAt, latest.CreatedAt}
if !releasedFrom.IsZero() {
query += "\nAND " + filterOpportunityReleasedFrom
args = append(args, releasedFrom.UTC())
}
query += orderOpportunitiesChanged
var rows []opportunityRow
err = s.db.SelectContext(ctx, &rows, query, args...)
if err != nil {
return nil, "", fmt.Errorf("listing changed opportunities: %w", err)
}
opportunities, err := s.opportunitiesFromRows(ctx, rows)
if err != nil {
return nil, "", err
}
return opportunities, latest.Hash, nil
}
// opportunitiesFromRows returns the opportunities in rows, in order, with
// their classifications and marks. The result is never nil.
func (s *Store) opportunitiesFromRows(
ctx context.Context, rows []opportunityRow,
) ([]types.Opportunity, error) {
opportunities := make([]types.Opportunity, len(rows))
for i, row := range rows {
opportunities[i] = row.Opportunity
}
if err := s.loadClassifications(ctx, opportunities); err != nil {
return nil, err
}
if err := s.loadMarks(ctx, opportunities); err != nil {
return nil, err
}
return opportunities, nil
}
// listOpportunitiesQuery assembles the ListOpportunities query from the
// fragments in queries.go, with one argument per filter that is set (none for
// mark filters).
func listOpportunitiesQuery(filter OpportunityFilter) (string, []any) {
var where []string
var args []any
add := func(clause string, argument any) {
where = append(where, clause)
args = append(args, argument)
}
if filter.Status != "" {
add(filterOpportunityStatus, filter.Status)
}
if filter.ClassificationCode != "" {
add(filterOpportunityClassificationCode, filter.ClassificationCode)
}
if match := ftsQuery(filter.Title); match != "" {
add(filterOpportunityTitleMatch, "title : ("+match+")")
}
if !filter.ReleasedFrom.IsZero() {
add(filterOpportunityReleasedFrom, filter.ReleasedFrom.UTC())
}
if !filter.ReleasedBefore.IsZero() {
add(filterOpportunityReleasedBefore, filter.ReleasedBefore.UTC())
}
marks := []struct {
filter MarkFilter
clause string
}{
{filter: filter.Hidden, clause: filterOpportunityMarkHidden},
{filter: filter.Skipped, clause: filterOpportunityMarkSkipped},
{filter: filter.Interested, clause: filterOpportunityMarkInterested},
{filter: filter.BidSubmitted, clause: filterOpportunityMarkBidSubmitted},
{filter: filter.Won, clause: filterOpportunityMarkWon},
{filter: filter.Lost, clause: filterOpportunityMarkLost},
}
for _, mark := range marks {
switch mark.filter {
case MarkSet:
where = append(where, mark.clause)
case MarkUnset:
where = append(where, "NOT "+mark.clause)
}
}
query := listOpportunitiesSelect
if len(where) > 0 {
query += "\nWHERE " + strings.Join(where, "\n\tAND ")
}
query += listOpportunitiesOrderAndPage
return query, append(args, filter.limit(), filter.Offset)
}
// loadClassifications fills in Classifications for all opportunities with one
// query.
func (s *Store) loadClassifications(
ctx context.Context, opportunities []types.Opportunity,
) error {
if len(opportunities) == 0 {
return nil
}
uuids := make([]string, len(opportunities))
byUUID := make(map[string]*types.Opportunity, len(opportunities))
for i := range opportunities {
uuids[i] = opportunities[i].UUID
byUUID[opportunities[i].UUID] = &opportunities[i]
}
query, args, err := sqlx.In(selectClassificationsForOpportunities, uuids)
if err != nil {
return err
}
// opportunity_uuid says which opportunity each row belongs to.
var rows []struct {
OpportunityUUID string `db:"opportunity_uuid"`
types.ClassificationCode
}
err = s.db.SelectContext(ctx, &rows, s.db.Rebind(query), args...)
if err != nil {
return err
}
for _, classificationRow := range rows {
opportunity := byUUID[classificationRow.OpportunityUUID]
opportunity.Classifications = append(
opportunity.Classifications, classificationRow.ClassificationCode,
)
}
return nil
}
// loadMarks fills in Marks for all opportunities with one query. An
// opportunity with no marks row gets zero marks carrying its UUID.
func (s *Store) loadMarks(
ctx context.Context, opportunities []types.Opportunity,
) error {
if len(opportunities) == 0 {
return nil
}
uuids := make([]string, len(opportunities))
byUUID := make(map[string]*types.Opportunity, len(opportunities))
for i := range opportunities {
uuids[i] = opportunities[i].UUID
byUUID[opportunities[i].UUID] = &opportunities[i]
opportunities[i].Marks = types.OpportunityMarks{
UUID: opportunities[i].UUID,
}
}
query, args, err := sqlx.In(selectMarksForOpportunities, uuids)
if err != nil {
return err
}
var rows []types.OpportunityMarks
err = s.db.SelectContext(ctx, &rows, s.db.Rebind(query), args...)
if err != nil {
return err
}
for _, row := range rows {
byUUID[row.UUID].Marks = row
}
return nil
}
// ListReleases returns a opportunity's releases with their tags, oldest first.
func (s *Store) ListReleases(
ctx context.Context, uuid string,
) ([]types.OpportunityRelease, error) {
releases := []types.OpportunityRelease{}
err := s.db.SelectContext(ctx, &releases, selectReleasesByOpportunity, uuid)
if err != nil {
return nil, err
}
byID := make(map[string]*types.OpportunityRelease, len(releases))
for i := range releases {
byID[releases[i].ReleaseID] = &releases[i]
}
var tags []types.OpportunityReleaseTag
err = s.db.SelectContext(ctx, &tags, selectReleaseTagsByOpportunity, uuid)
if err != nil {
return nil, err
}
for _, tag := range tags {
if release := byID[tag.ReleaseID]; release != nil {
release.Tags = append(release.Tags, tag)
}
}
return releases, nil
}
// ListClassifications returns every classification code that at least one
// stored opportunity has, ordered by code. A code no opportunity has any more
// isn't listed. The result is never nil.
func (s *Store) ListClassifications(
ctx context.Context,
) ([]types.ClassificationCode, error) {
classifications := []types.ClassificationCode{}
err := s.db.SelectContext(ctx, &classifications, selectClassificationsInUse)
if err != nil {
return nil, fmt.Errorf("listing classifications: %w", err)
}
return classifications, nil
}
// LatestReleaseDate returns the date of the newest release stored for the
// opportunities from source, or the zero time if none are stored, so a
// fetcher can carry on from where it left off.
func (s *Store) LatestReleaseDate(
ctx context.Context, source string,
) (time.Time, error) {
var latest time.Time
err := s.db.GetContext(ctx, &latest, selectSourceLatestReleaseDate, source)
if errors.Is(err, sql.ErrNoRows) {
return time.Time{}, nil
}
if err != nil {
return time.Time{}, fmt.Errorf(
"reading the latest release date of %s: %w", source, err,
)
}
return latest, nil
}
// OpportunityTitle is an opportunity's UUID, title and latest release date,
// as ListOtherSourcesTitles lists them.
type OpportunityTitle struct {
UUID string `db:"uuid"`
Title string `db:"title"`
LatestReleaseDate time.Time `db:"latest_release_date"`
}
// ListOtherSourcesTitles returns the UUID, title and latest release date of
// every opportunity not from source whose latest release is from from
// (inclusive) to before (exclusive), so a fetcher can find the ones a new
// opportunity repeats.
func (s *Store) ListOtherSourcesTitles(
ctx context.Context, source string, from, before time.Time,
) ([]OpportunityTitle, error) {
titles := []OpportunityTitle{}
err := s.db.SelectContext(
ctx,
&titles,
selectOtherSourcesTitles,
source,
from.UTC(),
before.UTC(),
)
if err != nil {
return nil, fmt.Errorf("listing titles of sources but %s: %w", source, err)
}
return titles, nil
}
// ReleaseExists reports whether a release is already stored, so a fetcher
// can skip converting it.
func (s *Store) ReleaseExists(
ctx context.Context, uuid, releaseID string,
) (bool, error) {
var exists bool
err := s.db.GetContext(ctx, &exists, selectReleaseExists, uuid, releaseID)
return exists, err
}