// 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" "database/sql" "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") // 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 } // 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 } // 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) 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"), }, 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 } // 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). // // 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 } 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. 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 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++ } } if err := tx.Commit(); err != nil { return SaveMultipleResult{}, err } return total, nil } // 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. 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) } _, err := s.db.NamedExecContext(ctx, upsertOpportunityMarks, marks) if isForeignKeyError(err) { return ErrNotFound } return err } // 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 } 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 } // 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 }