job-search/internal/mock/findatender/store.go

package mock

import (
	"bufio"
	"bytes"
	"encoding/json"
	"errors"
	"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(filepath.Clean(path))
	if os.IsNotExist(err) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	defer in.Close() //nolint:errcheck // read only

	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(filepath.Clean(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 {
				return nil, errors.Join(fmt.Errorf("repairing %s: %w", path, err), f.Close())
			}
		}
	}
	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()
}

// append stores r, syncing the file to disk if sync is true. 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
}

// Len returns the number of stored releases.
func (s *Store) Len() int {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return len(s.releases)
}

// HasOCID reports whether any release of the procedure ocid is stored.
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
}

// Close closes the store's file. The store must not be used afterwards.
func (s *Store) Close() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.f.Close()
}