1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
|
// runs.jsonl read/write: the serve-side run log. One JSON RunRow appended
// per non-empty poll batch; kept items double as the RSS render source, so
// each carries everything a feed item needs (title/link/date/cleaned
// content). Rotation is generous: past maxRunsBytes the file is rewritten
// keeping the newest rows, via temp + rename so a crash mid-rotate can't
// leave a torn log behind.
package core
import (
"bufio"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"strings"
"time"
)
// Kept-item kinds. Short strings so runs.jsonl stays human-greppable.
const (
KindRanked = "ranked"
KindRunnerUp = "runnerup"
KindRandom = "random"
)
// Rotation knobs. 20MB of runs is years of batches at expected sizes, and
// 500 retained rows is still far more than the count-based RSS tail (~21)
// will ever render.
const (
maxRunsBytes = 20 << 20
keepRunsOnRotate = 500
)
// KeptItem is one entry selected into a digest batch, with everything the
// RSS renderer needs so rendering never touches Miniflux.
type KeptItem struct {
EntryID int64 `json:"entry_id"`
URL string `json:"url"`
Title string `json:"title"`
FeedTitle string `json:"feed_title"`
PublishedAt time.Time `json:"published_at"`
Content string `json:"content"` // cleaned at poll time: HTML-stripped, <=5KB
Score float64 `json:"score"`
Kind string `json:"kind"` // KindRanked | KindRunnerUp | KindRandom
}
// RunRow is one poll batch: what was seen, what was kept, when, and how
// old the model was. EntriesSeen counts every fetched entry including any
// skipped before scoring (see RecFeedSkipped). ScoreMedian is the median
// score across every scored entry this poll — the batch's background
// calibration, rendered next to kept items' scores in the RSS body. It's
// a pointer so old rows (and empty polls) read as "absent", never as
// a misleading 0.00.
type RunRow struct {
RanAt time.Time `json:"ran_at"`
EntriesSeen int `json:"entries_seen"`
RecFeedSkipped int `json:"rec_feed_skipped,omitempty"`
Kept []KeptItem `json:"kept"`
DurationMs int64 `json:"duration_ms"`
ModelCreatedAt time.Time `json:"model_created_at"`
ScoreMedian *float64 `json:"score_median,omitempty"`
}
// ReadRuns parses all of path strictly. A missing file is not an error —
// it just means no runs logged yet (first serve boot).
func ReadRuns(path string) ([]RunRow, error) {
f, err := os.Open(path)
if err != nil {
if errors.Is(err, os.ErrNotExist) {
return nil, nil
}
return nil, fmt.Errorf("opening runs: %w", err)
}
defer func() { _ = f.Close() }()
var rows []RunRow
scanner := bufio.NewScanner(f)
scanner.Buffer(make([]byte, 64*1024), 4*1024*1024)
line := 0
for scanner.Scan() {
line++
text := strings.TrimSpace(scanner.Text())
if text == "" {
continue
}
var row RunRow
if err := json.Unmarshal([]byte(text), &row); err != nil {
return nil, fmt.Errorf("%s:%d: invalid JSON: %w", path, line, err)
}
if row.RanAt.IsZero() {
return nil, fmt.Errorf("%s:%d: ran_at is required", path, line)
}
for i, k := range row.Kept {
if k.EntryID <= 0 {
return nil, fmt.Errorf("%s:%d: kept[%d]: entry_id must be positive", path, line, i)
}
if k.Kind != KindRanked && k.Kind != KindRunnerUp && k.Kind != KindRandom {
return nil, fmt.Errorf("%s:%d: kept[%d]: invalid kind %q", path, line, i, k.Kind)
}
}
rows = append(rows, row)
}
if err := scanner.Err(); err != nil {
return nil, fmt.Errorf("reading %s: %w", path, err)
}
return rows, nil
}
// ReadRunsTail returns the newest k runs in file order (oldest to newest).
// k <= 0 reads everything.
func ReadRunsTail(path string, k int) ([]RunRow, error) {
rows, err := ReadRuns(path)
if err != nil {
return nil, err
}
if k > 0 && len(rows) > k {
rows = rows[len(rows)-k:]
}
return rows, nil
}
// AppendRun appends one row, then rotates the file if it has grown past
// maxRunsBytes. Appending is the only mutation; existing rows are never
// edited, only dropped by rotation.
func AppendRun(path string, row RunRow) error {
f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
if err != nil {
return fmt.Errorf("opening runs for append: %w", err)
}
if err := json.NewEncoder(f).Encode(row); err != nil {
_ = f.Close()
return fmt.Errorf("writing run row: %w", err)
}
if err := f.Close(); err != nil {
return fmt.Errorf("closing runs: %w", err)
}
info, err := os.Stat(path)
if err != nil {
return fmt.Errorf("stat runs: %w", err)
}
if info.Size() <= maxRunsBytes {
return nil
}
rows, err := ReadRuns(path)
if err != nil {
return err
}
if len(rows) > keepRunsOnRotate {
rows = rows[len(rows)-keepRunsOnRotate:]
}
tmp := filepath.Clean(path) + ".tmp"
w, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644)
if err != nil {
return fmt.Errorf("rotating runs: %w", err)
}
enc := json.NewEncoder(w)
for _, r := range rows {
if err := enc.Encode(r); err != nil {
_ = w.Close()
return fmt.Errorf("rotating runs: %w", err)
}
}
if err := w.Close(); err != nil {
return fmt.Errorf("rotating runs: %w", err)
}
if err := os.Rename(tmp, path); err != nil {
return fmt.Errorf("rotating runs: %w", err)
}
return nil
}
|
