| 1 | package apperr |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "sort" |
| 6 | "time" |
| 7 | |
| 8 | "github.com/carlosframework/platform/internal/applog" |
| 9 | "github.com/carlosframework/platform/internal/release" |
| 10 | ) |
| 11 | |
| 12 | // Prefix is the app's errors/ subtree (spec §3.2): a sibling family of |
| 13 | // logs/ with the same key shape below it. |
| 14 | // An invalid account or app is refused (applog.ErrBadKeySegment). |
| 15 | func Prefix(acct, app string) (string, error) { |
| 16 | return applog.FamilyPrefix(applog.FamilyErrors, acct, app) |
| 17 | } |
| 18 | |
| 19 | // BoxPrefix is one box's subtree within an app's errors. |
| 20 | func BoxPrefix(acct, app, box string) (string, error) { |
| 21 | return applog.FamilyBoxPrefix(applog.FamilyErrors, acct, app, box) |
| 22 | } |
| 23 | |
| 24 | // ChunkKey names one flushed errors chunk; see applog.ChunkKey. |
| 25 | func ChunkKey(acct, app, box, hour string, seq uint64, size int) (string, error) { |
| 26 | return applog.FamilyChunkKey(applog.FamilyErrors, acct, app, box, hour, seq, size) |
| 27 | } |
| 28 | |
| 29 | // HourOf is applog.HourOf, re-exported so shippers need one import. |
| 30 | func HourOf(t time.Time) string { return applog.HourOf(t) } |
| 31 | |
| 32 | // EncodeChunk renders records as gzipped JSONL. |
| 33 | func EncodeChunk(recs []Record) ([]byte, error) { return applog.EncodeJSONL(recs) } |
| 34 | |
| 35 | // DecodeChunk parses a chunk, skipping malformed lines. |
| 36 | func DecodeChunk(b []byte) ([]Record, error) { return applog.DecodeJSONL[Record](b) } |
| 37 | |
| 38 | // Filter is the read-side match. A coalesced record belongs to a window |
| 39 | // when [First, Last] overlaps [Since, Until] and is counted whole (spec |
| 40 | // §3.2: a window edge is wrong by at most one flush interval). |
| 41 | type Filter struct { |
| 42 | Since, Until time.Time |
| 43 | Host string |
| 44 | Source string |
| 45 | Release string |
| 46 | FP string |
| 47 | } |
| 48 | |
| 49 | // Match reports whether r survives f. |
| 50 | func (f Filter) Match(r Record) bool { |
| 51 | last, first := r.Last, r.First |
| 52 | if last.IsZero() { |
| 53 | last = r.TS |
| 54 | } |
| 55 | if first.IsZero() { |
| 56 | first = r.TS |
| 57 | } |
| 58 | if !f.Since.IsZero() && last.Before(f.Since) { |
| 59 | return false |
| 60 | } |
| 61 | if !f.Until.IsZero() && first.After(f.Until) { |
| 62 | return false |
| 63 | } |
| 64 | if f.Host != "" && r.Host != f.Host { |
| 65 | return false |
| 66 | } |
| 67 | if f.Source != "" && r.Source != f.Source { |
| 68 | return false |
| 69 | } |
| 70 | if f.Release != "" && r.Release != f.Release { |
| 71 | return false |
| 72 | } |
| 73 | if f.FP != "" && r.FP != f.FP { |
| 74 | return false |
| 75 | } |
| 76 | return true |
| 77 | } |
| 78 | |
| 79 | // ReadWindow assembles one app's errors/ records across boxes: every hour |
| 80 | // intersecting the window, filtered through f, merged by (Last, Box, Seq). |
| 81 | // collectCap bounds memory; past it the oldest go and the bool says so. |
| 82 | func ReadWindow(ctx context.Context, s release.Store, acct, app string, f Filter, collectCap int) ([]Record, bool, error) { |
| 83 | // A record's chunk sits under the hour of its FIRST occurrence (TS), |
| 84 | // but its Last can be much later — a group that started before the |
| 85 | // window opened and is still being added to at read time. ReadChunks |
| 86 | // prunes hour directories at listing time using the SAME since/until |
| 87 | // it is given, so listing with the exact f.Since would skip that |
| 88 | // chunk's hour entirely. Widen the LISTING window by one hour behind |
| 89 | // f.Since; f.Match (which tests [First, Last] against the true |
| 90 | // window) still does the exact filtering once the chunk is read. |
| 91 | listSince := f.Since |
| 92 | if !listSince.IsZero() { |
| 93 | listSince = listSince.Add(-time.Hour) |
| 94 | } |
| 95 | prefix, err := Prefix(acct, app) |
| 96 | if err != nil { |
| 97 | return nil, false, err |
| 98 | } |
| 99 | perBox, trimmed, err := applog.ReadChunks[Record](ctx, s, prefix, listSince, f.Until, f.Match, collectCap) |
| 100 | if err != nil { |
| 101 | return nil, false, err |
| 102 | } |
| 103 | merged := MergeRecords(perBox...) |
| 104 | if over := len(merged) - collectCap; over > 0 { |
| 105 | merged = merged[over:] |
| 106 | trimmed = true |
| 107 | } |
| 108 | return merged, trimmed, nil |
| 109 | } |
| 110 | |
| 111 | // MergeRecords folds per-source slices into one timeline ordered by |
| 112 | // (Last, Box, Seq), oldest first. |
| 113 | func MergeRecords(slices ...[]Record) []Record { |
| 114 | var out []Record |
| 115 | for _, s := range slices { |
| 116 | out = append(out, s...) |
| 117 | } |
| 118 | sort.SliceStable(out, func(i, j int) bool { |
| 119 | a, b := out[i], out[j] |
| 120 | if !a.Last.Equal(b.Last) { |
| 121 | return a.Last.Before(b.Last) |
| 122 | } |
| 123 | if a.Box != b.Box { |
| 124 | return a.Box < b.Box |
| 125 | } |
| 126 | return a.Seq < b.Seq |
| 127 | }) |
| 128 | return out |
| 129 | } |
| 130 | |
amadan