//go:build !js && !wasip1
package obsduck
import (
"context"
"fmt"
"time"
"craftdesign.group/quarry/internal/obs"
)
// flushRollup writes the hours that have closed, or every hour when all
// is set (shutdown). A failed write puts the counts back for the next
// try and is remembered like any other.
func (s *Sink) flushRollup(all bool) {
var before time.Time
if !all {
before = time.Now().UTC().Truncate(time.Hour)
}
rows := s.roll.Take(before)
if len(rows) == 0 {
return
}
if err := s.writeHourly(s.ctx, rows); err != nil {
s.roll.Restore(rows)
s.mu.Lock()
s.flushErr = err
s.mu.Unlock()
}
}
func (s *Sink) writeHourly(ctx context.Context, rows []obs.HourCount) error {
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
st, err := tx.Prepare(`INSERT INTO hit_hourly
(hour, pattern, signed_in, agent, network, country, requests, elapsed_ms)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`)
if err != nil {
return err
}
defer st.Close()
for _, c := range rows {
if _, err := st.ExecContext(ctx, c.Hour, c.Pattern, c.SignedIn, c.Agent,
c.Network, c.Country, c.Requests, c.Elapsed.Milliseconds()); err != nil {
return err
}
}
return tx.Commit()
}
// Hourly reads the closed hours back, summed per key, oldest first. It
// is what the operations page asks to compare a route's rate with its
// own norm; the hour still open is not in it.
func (s *Sink) Hourly(q obs.HourlyQuery) ([]obs.HourCount, error) {
s.Flush()
query := `SELECT hour, pattern, signed_in, agent, coalesce(network, ''), coalesce(country, ''),
sum(requests)::BIGINT, sum(elapsed_ms)::BIGINT FROM hit_hourly WHERE 1 = 1`
var args []any
if q.Pattern != "" {
query += ` AND pattern = ?`
args = append(args, q.Pattern)
}
if !q.Since.IsZero() {
query += ` AND hour >= ?`
args = append(args, q.Since.UTC())
}
if !q.Until.IsZero() {
query += ` AND hour < ?`
args = append(args, q.Until.UTC())
}
query += ` GROUP BY ALL ORDER BY hour, pattern, signed_in, agent, 5, 6`
rows, err := s.db.QueryContext(s.ctx, query, args...)
if err != nil {
return nil, fmt.Errorf("obsduck: hourly: %w", err)
}
defer rows.Close()
var out []obs.HourCount
for rows.Next() {
var c obs.HourCount
var ms int64
if err := rows.Scan(&c.Hour, &c.Pattern, &c.SignedIn, &c.Agent, &c.Network, &c.Country, &c.Requests, &ms); err != nil {
return nil, err
}
c.Hour = c.Hour.UTC()
c.Elapsed = time.Duration(ms) * time.Millisecond
out = append(out, c)
}
return out, rows.Err()
}