package mock import ( "bufio" "bytes" "encoding/json" "fmt" "log" "math/rand" "os" "path/filepath" "sort" "strconv" "strings" "sync" "time" "git.avsdev.uk/Go/job-search/internal/mock/findATender/ocds" ) // Store keeps every release in memory and appends each new one to a JSON Lines // file, so releases survive restarts. Releases are kept in date order. type Store struct { mu sync.RWMutex f *os.File releases []ocds.Release times []time.Time latest map[string]int // ocid -> index of its most recent release lastNum map[int]int // year -> highest notice number used // Procedures not yet awarded, as a slice plus positions so that adding, // removing and picking one at random are all constant time. open []string openPos map[string]int } // OpenStore loads the releases stored at path, creating the file and its // directory if needed, and opens it for appending. Lines that can't be read // are skipped, releases are put back in date order if needed, and a record // left incomplete by a crash is ended so new ones start on a fresh line. func OpenStore(path string) (*Store, error) { s := &Store{latest: map[string]int{}, lastNum: map[int]int{}, openPos: map[string]int{}} if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil { return nil, err } releases, err := readReleases(path) if err != nil { return nil, err } // The file is written in order, but sort anyway in case it was hand-edited. sort.SliceStable(releases, func(i, j int) bool { return releases[i].Date.Before(releases[j].Date) }) for _, r := range releases { s.releases = append(s.releases, r) s.times = append(s.times, r.Date) } for i, r := range s.releases { s.index(i, r) } if s.f, err = openForAppend(path); err != nil { return nil, err } return s, nil } // readReleases reads the releases stored at path, skipping lines that are // blank or can't be decoded. A missing file holds no releases. func readReleases(path string) ([]ocds.Release, error) { in, err := os.Open(path) if os.IsNotExist(err) { return nil, nil } if err != nil { return nil, err } defer in.Close() var releases []ocds.Release sc := bufio.NewScanner(in) sc.Buffer(make([]byte, 1<<20), 64<<20) for line := 1; sc.Scan(); line++ { b := bytes.TrimSpace(sc.Bytes()) if len(b) == 0 { continue } var r ocds.Release if err := json.Unmarshal(b, &r); err != nil { log.Printf("store: skipping unreadable line %d: %v", line, err) continue } releases = append(releases, r) } if err := sc.Err(); err != nil { return nil, fmt.Errorf("reading %s: %w", path, err) } return releases, nil } // openForAppend opens path for appending releases. If a previous run died // mid-write, it ends the partial record so the next one starts on a fresh // line. func openForAppend(path string) (*os.File, error) { f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0o600) if err != nil { return nil, err } if st, err := f.Stat(); err == nil && st.Size() > 0 { last := make([]byte, 1) if _, err := f.ReadAt(last, st.Size()-1); err == nil && last[0] != '\n' { if _, err := f.Write([]byte("\n")); err != nil { f.Close() return nil, fmt.Errorf("repairing %s: %w", path, err) } } } return f, nil } // index records release r, at position i, in the lookups: the ocid's latest // release, whether its procedure is still open (an award closes it), and the // highest notice number used each year. func (s *Store) index(i int, r ocds.Release) { s.latest[r.OCID] = i if len(r.Tag) > 0 && r.Tag[0] == "award" { s.closeProcedure(r.OCID) } else if _, ok := s.openPos[r.OCID]; !ok { s.openPos[r.OCID] = len(s.open) s.open = append(s.open, r.OCID) } if num, year, ok := strings.Cut(r.ID, "-"); ok { n, err1 := strconv.Atoi(num) y, err2 := strconv.Atoi(year) if err1 == nil && err2 == nil && n > s.lastNum[y] { s.lastNum[y] = n } } } // closeProcedure removes ocid from the open procedures, if it is there. func (s *Store) closeProcedure(ocid string) { pos, ok := s.openPos[ocid] if !ok { return } last := len(s.open) - 1 s.open[pos] = s.open[last] s.openPos[s.open[pos]] = pos s.open = s.open[:last] delete(s.openPos, ocid) } // Append persists a release and flushes it to disk. func (s *Store) Append(r ocds.Release) error { return s.append(r, true) } // AppendNoSync persists a release without an fsync; call Sync afterwards. // Used for bulk catch-up, where a sync per record would be very slow. func (s *Store) AppendNoSync(r ocds.Release) error { return s.append(r, false) } // Sync flushes appended releases to disk. Call it after AppendNoSync. func (s *Store) Sync() error { s.mu.Lock() defer s.mu.Unlock() return s.f.Sync() } // A date earlier than the newest stored release (for example after a clock // change) is moved up so the order stays valid. func (s *Store) append(r ocds.Release, sync bool) error { s.mu.Lock() defer s.mu.Unlock() r.Date = r.Date.UTC() if n := len(s.times); n > 0 && r.Date.Before(s.times[n-1]) { r.Date = s.times[n-1] } b, err := json.Marshal(r) if err != nil { return err } if _, err := s.f.Write(append(b, '\n')); err != nil { return err } if sync { if err := s.f.Sync(); err != nil { return err } } s.releases = append(s.releases, r) s.times = append(s.times, r.Date) s.index(len(s.releases)-1, r) return nil } // Page returns up to limit releases dated in [from, upper], starting no earlier // than position start, plus the position to continue from and whether more exist. func (s *Store) Page(from, upper time.Time, start, limit int) ([]ocds.Release, int, bool) { s.mu.RLock() defer s.mu.RUnlock() i := sort.Search(len(s.times), func(i int) bool { return !s.times[i].Before(from) }) if start > i { i = start } out := make([]ocds.Release, 0, limit) for i < len(s.releases) && len(out) < limit && !s.times[i].After(upper) { out = append(out, s.releases[i]) i++ } more := i < len(s.releases) && !s.times[i].After(upper) return out, i, more } // Last returns the date of the newest release, or false if the store is // empty. func (s *Store) Last() (time.Time, bool) { s.mu.RLock() defer s.mu.RUnlock() if len(s.times) == 0 { return time.Time{}, false } return s.times[len(s.times)-1], true } func (s *Store) Len() int { s.mu.RLock() defer s.mu.RUnlock() return len(s.releases) } func (s *Store) HasOCID(ocid string) bool { s.mu.RLock() defer s.mu.RUnlock() _, ok := s.latest[ocid] return ok } // NextNoticeID returns the next FTS-style notice id ("000123-2026") for year. func (s *Store) NextNoticeID(year int) string { s.mu.RLock() defer s.mu.RUnlock() return fmt.Sprintf("%06d-%d", s.lastNum[year]+1, year) } // RandomOpen picks a procedure that has not been awarded and whose latest // release is after since. Procedures idle since before then are dropped from // the open set, as if they had been abandoned. func (s *Store) RandomOpen(r *rand.Rand, since time.Time) (ocds.Release, bool) { s.mu.Lock() defer s.mu.Unlock() for len(s.open) > 0 { ocid := s.open[r.Intn(len(s.open))] i := s.latest[ocid] if s.times[i].After(since) { return s.releases[i], true } s.closeProcedure(ocid) } return ocds.Release{}, false } func (s *Store) Close() error { s.mu.Lock() defer s.mu.Unlock() return s.f.Close() }