| 1 | package analytics |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "os" |
| 9 | "strings" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/applog" |
| 12 | "github.com/carlosframework/platform/internal/release" |
| 13 | ) |
| 14 | |
| 15 | // HourRow is one UTC hour's counters. |
| 16 | type HourRow struct { |
| 17 | PV int64 `json:"pv"` |
| 18 | Bots int64 `json:"bots"` |
| 19 | Bytes int64 `json:"bytes"` |
| 20 | } |
| 21 | |
| 22 | // DropCounts is why events were dropped at a door (steps 3–4 fill them). |
| 23 | type DropCounts struct { |
| 24 | Rate int64 `json:"rate"` |
| 25 | Size int64 `json:"size"` |
| 26 | Origin int64 `json:"origin"` |
| 27 | Malformed int64 `json:"malformed"` |
| 28 | } |
| 29 | |
| 30 | // Dropped is the honesty block (spec §3.3). |
| 31 | type Dropped struct { |
| 32 | E DropCounts `json:"e"` |
| 33 | A DropCounts `json:"a"` |
| 34 | Flush int64 `json:"flush"` |
| 35 | } |
| 36 | |
| 37 | // Day is apps/<sqid>/<app>/analytics/<box>/days/<date>.json (spec §3.3): |
| 38 | // one box's counting for one app for one UTC day. Additive shape. |
| 39 | type Day struct { |
| 40 | V int `json:"v"` |
| 41 | Date string `json:"date"` |
| 42 | Hours map[string]HourRow `json:"hours"` |
| 43 | Visitors string `json:"visitors"` |
| 44 | Pages map[string]int64 `json:"pages"` |
| 45 | Referers map[string]int64 `json:"referers"` |
| 46 | Events map[string]int64 `json:"events"` |
| 47 | Seq map[string]int64 `json:"seq"` |
| 48 | Dropped Dropped `json:"dropped"` |
| 49 | Countries map[string]int64 `json:"countries"` |
| 50 | // GeoSources counts pageviews per database type that answered (geoip |
| 51 | // spec §9): the provenance the attribution on the analytics page needs. |
| 52 | GeoSources map[string]int64 `json:"geo_sources,omitempty"` |
| 53 | Restarted bool `json:"restarted,omitempty"` |
| 54 | } |
| 55 | |
| 56 | // Delta is one flush batch from one box for one app-day. |
| 57 | type Delta struct { |
| 58 | Date string |
| 59 | Box string |
| 60 | Seq int64 |
| 61 | Hours map[string]HourRow |
| 62 | Visitors *HLL |
| 63 | Pages *TopK |
| 64 | Referers *TopK |
| 65 | Events *TopK |
| 66 | Dropped Dropped |
| 67 | Restarted bool |
| 68 | // Countries and GeoSources are pageview counts by GeoIP answer (geoip |
| 69 | // spec §9), bounded by boundedAdd. |
| 70 | Countries map[string]int64 |
| 71 | GeoSources map[string]int64 |
| 72 | |
| 73 | // requeued marks a delta that has already been through one failed |
| 74 | // flush (internal/analytics.Meter.Requeue). Not serialised: it never |
| 75 | // reaches the day file, only decides whether a second Requeue keeps |
| 76 | // or drops the delta. |
| 77 | requeued bool |
| 78 | } |
| 79 | |
| 80 | // Prefix is the app's analytics/ subtree. An invalid account or app is |
| 81 | // refused (applog.ErrBadKeySegment), never built. |
| 82 | func Prefix(sqid, app string) (string, error) { |
| 83 | return applog.FamilyPrefix(applog.FamilyAnalytics, sqid, app) |
| 84 | } |
| 85 | |
| 86 | // BoxPrefix is one box's day files. |
| 87 | func BoxPrefix(sqid, app, box string) (string, error) { |
| 88 | p, err := applog.FamilyBoxPrefix(applog.FamilyAnalytics, sqid, app, box) |
| 89 | if err != nil { |
| 90 | return "", err |
| 91 | } |
| 92 | return p + "days/", nil |
| 93 | } |
| 94 | |
| 95 | // DayKey is one day file. |
| 96 | func DayKey(sqid, app, box, date string) (string, error) { |
| 97 | p, err := BoxPrefix(sqid, app, box) |
| 98 | if err != nil { |
| 99 | return "", err |
| 100 | } |
| 101 | if err := applog.CheckSegment("date", date); err != nil { |
| 102 | return "", err |
| 103 | } |
| 104 | return p + date + ".json", nil |
| 105 | } |
| 106 | |
| 107 | // Merge folds x into d unless x.Seq is not above the recorded seq for |
| 108 | // x.Box — the idempotency rule that makes a retried conditional put safe. |
| 109 | func (d *Day) Merge(x Delta) (bool, error) { |
| 110 | if x.Seq <= d.Seq[x.Box] { |
| 111 | return false, nil |
| 112 | } |
| 113 | if d.Seq == nil { |
| 114 | d.Seq = map[string]int64{} |
| 115 | } |
| 116 | d.V, d.Date = 1, x.Date |
| 117 | if d.Hours == nil { |
| 118 | d.Hours = map[string]HourRow{} |
| 119 | } |
| 120 | for h, row := range x.Hours { |
| 121 | cur := d.Hours[h] |
| 122 | cur.PV += row.PV |
| 123 | cur.Bots += row.Bots |
| 124 | cur.Bytes += row.Bytes |
| 125 | d.Hours[h] = cur |
| 126 | } |
| 127 | // A corrupt or unreadable sketch must not poison the whole day: start |
| 128 | // fresh rather than fail the merge. The bad string is overwritten by |
| 129 | // the freshly encoded sketch below either way. |
| 130 | vis, err := DecodeHLL(d.Visitors) |
| 131 | if err != nil { |
| 132 | vis = &HLL{} |
| 133 | } |
| 134 | if x.Visitors != nil { |
| 135 | vis.Merge(x.Visitors) |
| 136 | } |
| 137 | d.Visitors = vis.Encode() |
| 138 | d.Pages = mergeTop(d.Pages, x.Pages) |
| 139 | d.Referers = mergeTop(d.Referers, x.Referers) |
| 140 | d.Events = mergeTop(d.Events, x.Events) |
| 141 | d.Dropped.E = addDrops(d.Dropped.E, x.Dropped.E) |
| 142 | d.Dropped.A = addDrops(d.Dropped.A, x.Dropped.A) |
| 143 | d.Dropped.Flush += x.Dropped.Flush |
| 144 | d.Restarted = d.Restarted || x.Restarted |
| 145 | d.Countries = mergeCounts(d.Countries, x.Countries, maxCountryKeys, nil) |
| 146 | d.GeoSources = mergeCounts(d.GeoSources, x.GeoSources, maxGeoSourceKeys, attributed) |
| 147 | d.Seq[x.Box] = x.Seq |
| 148 | return true, nil |
| 149 | } |
| 150 | |
| 151 | func mergeTop(m map[string]int64, t *TopK) map[string]int64 { |
| 152 | k := TopKFrom(m) |
| 153 | if t != nil { |
| 154 | k.Merge(t) |
| 155 | } |
| 156 | return k.Fold() |
| 157 | } |
| 158 | |
| 159 | func addDrops(a, b DropCounts) DropCounts { |
| 160 | return DropCounts{a.Rate + b.Rate, a.Size + b.Size, a.Origin + b.Origin, a.Malformed + b.Malformed} |
| 161 | } |
| 162 | |
| 163 | // UpsertDay CAS-merges x into the app-day object for x.Box: read (etag), |
| 164 | // Merge, conditional put; on ErrCASConflict re-read and retry, three |
| 165 | // attempts. A no-op merge (seq already recorded) writes nothing. |
| 166 | func UpsertDay(ctx context.Context, s release.Store, sqid, app string, x Delta) error { |
| 167 | key, err := DayKey(sqid, app, x.Box, x.Date) |
| 168 | if err != nil { |
| 169 | return err |
| 170 | } |
| 171 | for attempt := 0; attempt < 3; attempt++ { |
| 172 | var day Day |
| 173 | raw, etag, err := s.Get(ctx, key) |
| 174 | switch { |
| 175 | case err == nil: |
| 176 | if uerr := json.Unmarshal(raw, &day); uerr != nil { |
| 177 | return fmt.Errorf("analytics: parse %s: %w", key, uerr) |
| 178 | } |
| 179 | case errors.Is(err, os.ErrNotExist): |
| 180 | etag = "" |
| 181 | default: |
| 182 | return err |
| 183 | } |
| 184 | applied, err := day.Merge(x) |
| 185 | if err != nil { |
| 186 | return err |
| 187 | } |
| 188 | if !applied { |
| 189 | return nil |
| 190 | } |
| 191 | out, err := json.Marshal(day) |
| 192 | if err != nil { |
| 193 | return err |
| 194 | } |
| 195 | if _, err := s.Put(ctx, key, out, etag); err == nil { |
| 196 | return nil |
| 197 | } else if !errors.Is(err, release.ErrCASConflict) { |
| 198 | return err |
| 199 | } |
| 200 | } |
| 201 | return fmt.Errorf("analytics: %s: lost CAS race three times", key) |
| 202 | } |
| 203 | |
| 204 | const ( |
| 205 | maxCountryKeys = 300 |
| 206 | maxGeoSourceKeys = 8 |
| 207 | ) |
| 208 | |
| 209 | // attributed reports whether a database type carries an attribution the |
| 210 | // analytics page must keep showing (geoip spec §9, §13: the DB-IP and |
| 211 | // GeoLite2 prefixes). Such a source is never folded into _other, so its |
| 212 | // credit survives any number of other types (codex R7-3). Kept here as two |
| 213 | // prefixes rather than importing internal/geoip, which this package's |
| 214 | // dependants should not need. |
| 215 | func attributed(source string) bool { |
| 216 | return strings.HasPrefix(source, "DBIP-") || strings.HasPrefix(source, "GeoLite2-") |
| 217 | } |
| 218 | |
| 219 | // boundedAdd adds n to m[k], folding a new key past max into OtherKey |
| 220 | // unless keep says it must never fold. |
| 221 | func boundedAdd(m map[string]int64, k string, n int64, max int, keep func(string) bool) { |
| 222 | if _, ok := m[k]; ok || k == OtherKey || len(m) < max || (keep != nil && keep(k)) { |
| 223 | m[k] += n |
| 224 | return |
| 225 | } |
| 226 | m[OtherKey] += n |
| 227 | } |
| 228 | |
| 229 | func mergeCounts(dst, src map[string]int64, max int, keep func(string) bool) map[string]int64 { |
| 230 | if dst == nil { |
| 231 | dst = map[string]int64{} |
| 232 | } |
| 233 | for k, n := range src { |
| 234 | boundedAdd(dst, k, n, max, keep) |
| 235 | } |
| 236 | return dst |
| 237 | } |
| 238 | |
| 239 | // countGeo applies spec §9 to one pageview: hit counts its country (or |
| 240 | // _unknown for a coordinate-only answer), miss counts _unknown, off and "" |
| 241 | // count nothing; a hit or miss also counts its source. |
| 242 | func countGeo(d *Delta, o Obs) { |
| 243 | if o.GeoState != "hit" && o.GeoState != "miss" { |
| 244 | return |
| 245 | } |
| 246 | c := o.Country |
| 247 | if o.GeoState == "miss" || c == "" { |
| 248 | c = "_unknown" |
| 249 | } |
| 250 | boundedAdd(d.Countries, c, 1, maxCountryKeys, nil) |
| 251 | if o.GeoSource != "" { |
| 252 | boundedAdd(d.GeoSources, o.GeoSource, 1, maxGeoSourceKeys, attributed) |
| 253 | } |
| 254 | } |
| 255 | |
amadan