// 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, 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 }