// 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" ) // ErrNotFound is returned when a requested opportunity does not exist. var ErrNotFound = errors.New("store: not found") // 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 { // 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 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 t 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(t *types.Opportunity) error { if t == nil { return errors.New("store: nil opportunity") } if t.UUID == "" { return errors.New("store: opportunity has no UUID") } t.Marks.UUID = t.UUID if err := t.Marks.Validate(); err != nil { return fmt.Errorf("store: opportunity %s: %w", t.UUID, err) } latest := t.LatestReleaseDate for i := range t.Releases { r := &t.Releases[i] if r.ReleaseID == "" { return fmt.Errorf("store: opportunity %s: release %d has no release id", t.UUID, i) } r.UUID = t.UUID r.ReleaseDate = r.ReleaseDate.UTC() if r.ReleaseDate.After(latest) { latest = r.ReleaseDate } for j := range r.Tags { r.Tags[j].UUID = t.UUID r.Tags[j].ReleaseID = r.ReleaseID } } if latest.IsZero() { return fmt.Errorf("store: opportunity %s has no release date", t.UUID) } t.LatestReleaseDate = latest.UTC() t.StartDate = utcPtr(t.StartDate) t.EndDate = utcPtr(t.EndDate) return nil } func utcPtr(t *time.Time) *time.Time { if t == nil { return nil } u := t.UTC() return &u } // 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(r rune) bool { return !unicode.IsLetter(r) && !unicode.IsNumber(r) }) for i, w := range words { words[i] = `"` + w + `"*` } 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 o options for _, opt := range opts { opt(&o) } 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, o); 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 t 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 t'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. // // t.Marks are stored only for a new opportunity; existing marks are kept (use // SetMarks to change them). // // t is normalised in place (see prepareOpportunity), and when the opportunity is // written its classifications' IDs are filled in. func (s *Store) SaveOpportunity(ctx context.Context, t *types.Opportunity) (SaveResult, error) { if err := prepareOpportunity(t); 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 res, _, err := saveOpportunity(ctx, tx, t) if err != nil { return SaveResult{}, err } if err := tx.Commit(); err != nil { return SaveResult{}, err } return res, nil } // SaveOpportunities stores t 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 t'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. // // t.Marks are stored only for a new opportunity; existing marks are kept (use // SetMarks to change them). // // t is normalised in place (see prepareOpportunity), and when the opportunity is // written its classifications' IDs are filled in. func (s *Store) SaveOpportunities( ctx context.Context, opportunities []*types.Opportunity, ) (SaveMultipleResult, error) { for _, t := range opportunities { if err := prepareOpportunity(t); 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 allRes := SaveMultipleResult{OpportunityResults: make([]SaveResult, len(opportunities))} for i, t := range opportunities { res, isNew, err := saveOpportunity(ctx, tx, t) if err != nil { return SaveMultipleResult{}, err } allRes.OpportunityResults[i] = res allRes.NewReleases += res.NewReleases switch { case isNew: allRes.NewOpportunities++ case res.OpportunityUpdated: allRes.UpdatedOpportunities++ } } if err := tx.Commit(); err != nil { return SaveMultipleResult{}, err } return allRes, 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, t *types.Opportunity, ) (res SaveResult, isNew bool, err error) { var stored time.Time err = tx.GetContext(ctx, &stored, selectOpportunityLatestReleaseDate, t.UUID) switch { case errors.Is(err, sql.ErrNoRows): isNew = true res.OpportunityUpdated = true case err != nil: return SaveResult{}, false, fmt.Errorf("store: reading opportunity %s: %w", t.UUID, err) default: res.OpportunityUpdated = !t.LatestReleaseDate.Before(stored) } if res.OpportunityUpdated { if err := writeOpportunity(ctx, tx, t); err != nil { return SaveResult{}, false, fmt.Errorf("store: writing opportunity %s: %w", t.UUID, err) } } for _, r := range t.Releases { inserted, err := insertReleaseWithTags(ctx, tx, r) if err != nil { return SaveResult{}, false, err } if inserted { res.NewReleases++ } } return res, isNew, nil } // insertReleaseWithTags stores r and its tags unless r is already stored, and // reports whether it was inserted. func insertReleaseWithTags( ctx context.Context, tx *sqlx.Tx, r types.OpportunityRelease, ) (bool, error) { result, err := tx.NamedExecContext(ctx, insertRelease, r) if err != nil { return false, fmt.Errorf("store: inserting release %s/%s: %w", r.UUID, r.ReleaseID, err) } if n, err := result.RowsAffected(); err != nil || n == 0 { return false, err // n == 0: already stored } for _, tag := range r.Tags { if _, err := tx.NamedExecContext(ctx, insertReleaseTag, tag); err != nil { return false, fmt.Errorf( "store: inserting tag for %s/%s: %w", r.UUID, r.ReleaseID, err, ) } } return true, nil } // writeOpportunity upserts t'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, t *types.Opportunity) (err error) { if _, err := tx.NamedExecContext(ctx, upsertOpportunity, t); err != nil { return err } if _, err := tx.NamedExecContext(ctx, insertOpportunityMarks, t.Marks); err != nil { return err } if _, err := tx.ExecContext(ctx, deleteOpportunityClassifications, t.UUID); err != nil { return err } stmt, err := tx.PrepareNamedContext(ctx, upsertClassificationCodeReturningID) if err != nil { return err } defer func() { err = errors.Join(err, stmt.Close()) }() for i := range t.Classifications { c := &t.Classifications[i] if err := stmt.GetContext(ctx, &c.ID, c); err != nil { return err } if _, err := tx.ExecContext(ctx, insertOpportunityClassification, t.UUID, c.ID); err != nil { return err } } return nil } // SetMarks replaces the marks of the opportunity m.UUID, or returns ErrNotFound. // Marks that fail m.Validate (e.g. types.ErrWonAndLost) are not stored. func (s *Store) SetMarks(ctx context.Context, m types.OpportunityMarks) error { if m.UUID == "" { return errors.New("store: marks have no UUID") } if err := m.Validate(); err != nil { return fmt.Errorf("store: opportunity %s: %w", m.UUID, err) } _, err := s.db.NamedExecContext(ctx, upsertOpportunityMarks, m) 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 se sqlite3.Error return errors.As(err, &se) && se.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 f, with their classifications // and marks but without releases (use ListReleases or GetOpportunity for those). func (s *Store) ListOpportunities( ctx context.Context, f OpportunityFilter, ) ([]types.Opportunity, error) { query, args := listOpportunitiesQuery(f) 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(f OpportunityFilter) (string, []any) { var where []string var args []any add := func(clause string, arg any) { where = append(where, clause) args = append(args, arg) } if f.Status != "" { add(filterOpportunityStatus, f.Status) } if f.ClassificationCode != "" { add(filterOpportunityClassificationCode, f.ClassificationCode) } if q := ftsQuery(f.Title); q != "" { add(filterOpportunityTitleMatch, q) } if !f.ReleasedFrom.IsZero() { add(filterOpportunityReleasedFrom, f.ReleasedFrom.UTC()) } if !f.ReleasedBefore.IsZero() { add(filterOpportunityReleasedBefore, f.ReleasedBefore.UTC()) } for _, m := range []struct { filter MarkFilter clause string }{ {f.Hidden, filterOpportunityMarkHidden}, {f.Skipped, filterOpportunityMarkSkipped}, {f.Interested, filterOpportunityMarkInterested}, {f.BidSubmitted, filterOpportunityMarkBidSubmitted}, {f.Won, filterOpportunityMarkWon}, {f.Lost, filterOpportunityMarkLost}, } { switch m.filter { case MarkSet: where = append(where, m.clause) case MarkUnset: where = append(where, "NOT "+m.clause) } } query := listOpportunitiesSelect if len(where) > 0 { query += "\nWHERE " + strings.Join(where, "\n\tAND ") } query += listOpportunitiesOrderAndPage return query, append(args, f.limit(), f.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 } if err := s.db.SelectContext(ctx, &rows, s.db.Rebind(query), args...); err != nil { return err } for _, r := range rows { t := byUUID[r.OpportunityUUID] t.Classifications = append(t.Classifications, r.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 if err := s.db.SelectContext(ctx, &rows, s.db.Rebind(query), args...); err != nil { return err } for _, m := range rows { byUUID[m.UUID].Marks = m } 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{} if err := s.db.SelectContext(ctx, &releases, selectReleasesByOpportunity, uuid); 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 if err := s.db.SelectContext(ctx, &tags, selectReleaseTagsByOpportunity, uuid); err != nil { return nil, err } for _, tag := range tags { if r := byID[tag.ReleaseID]; r != nil { r.Tags = append(r.Tags, tag) } } return releases, 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 }