Skip to content
quarry Artifact [df61923673]
Login

Artifact [df61923673]

Check-ins Using Download Hex

Artifact df61923673c18ac1e8a7931859c83b3df1dfb0812eba99ae8967cbe32f5a81d4:

//go:build !js && !wasip1

// Package obsduck stores observations for analysis.
//
// It is separate from internal/obs, and behind a build constraint, for
// one reason: DuckDB requires cgo and refuses to build for wasm. Both
// were measured rather than assumed — CGO_ENABLED=0 fails on
// go-duckdb's bindings, and GOOS=js reports "build constraints exclude
// all Go files". quarry is otherwise CGO-free on purpose (its SQLite
// driver is a wasm build for exactly this reason), so that the forge
// can theoretically compile to wasm for an embedded UI. That build
// ships no analytics, wires no sink, and must never reach this package.
//
// Nothing on the request path may import this. obs.Sink is the seam;
// only the composition root knows which implementation exists.
//
// Columnar because the workload is columnar: append-only, time-ordered,
// aggregate-queried, never point-looked-up by a page. RFC 0001 defers
// the lake to "when hydration and analytics — the things columnar is
// for — exist", and reserves the operational surface for SQLite. This
// is that analytics workload, not a second operational store.
package obsduck

import (
	"database/sql"
	"fmt"
	"strings"
	"sync"
	"time"

	_ "github.com/marcboeker/go-duckdb/v2"

	"craftdesign.group/quarry/internal/obs"
)

// Sink writes observations to a DuckDB file.
//
// Buffered, never a write per request: an analytics store on the hot
// path would be the forge paying at serve time for a question nobody is
// asking yet. Rows accumulate in memory and land in batches, which is
// also the shape a Parquet COPY TO wants when the lake arrives.
type Sink struct {
	db *sql.DB

	mu       sync.Mutex
	renders  []obs.Render
	subs     []obs.Subprocess
	batch    int
	closed   bool
	flushErr error
}

// Open creates or opens the analytics store at path.
func Open(path string, batch int) (*Sink, error) {
	if batch <= 0 {
		batch = 256
	}
	db, err := sql.Open("duckdb", path)
	if err != nil {
		return nil, fmt.Errorf("obsduck: open %s: %w", path, err)
	}
	if err := db.Ping(); err != nil {
		db.Close()
		return nil, fmt.Errorf("obsduck: open %s: %w", path, err)
	}
	// Wide, flat and denormalised, which is what columnar scans want and
	// what a Parquet export wants. No indexes and no keys: nothing here
	// is looked up by identity, and a row is never updated.
	for _, ddl := range []string{
		// ts, not "at": AT is reserved in DuckDB (AT TIME ZONE, time
		// travel), and a quoted identifier would need quoting at every
		// call site and in every ad-hoc query someone writes later.
		`CREATE TABLE IF NOT EXISTS render (
			ts TIMESTAMP, path VARCHAR, repo VARCHAR,
			elapsed_ms BIGINT, ops_ms BIGINT, zit_ms BIGINT, status INTEGER)`,
		// url is the path actually requested, added after the fact: the
		// route groups rows and this identifies one. Additive so an
		// existing store keeps its history with the column empty,
		// rather than a schema bump that would throw the history away
		// to gain a column describing it.
		`ALTER TABLE render ADD COLUMN IF NOT EXISTS url VARCHAR`,
		`CREATE TABLE IF NOT EXISTS subprocess (
			ts TIMESTAMP, verb VARCHAR, repo VARCHAR,
			duration_ms BIGINT, failed BOOLEAN)`,
	} {
		if _, err := db.Exec(ddl); err != nil {
			db.Close()
			return nil, fmt.Errorf("obsduck: schema: %w", err)
		}
	}
	return &Sink{db: db, batch: batch}, nil
}

// Render buffers one page.
func (s *Sink) Render(v obs.Render) {
	s.mu.Lock()
	s.renders = append(s.renders, v)
	full := len(s.renders) >= s.batch
	s.mu.Unlock()
	if full {
		s.Flush()
	}
}

// Subprocess buffers one store call.
func (s *Sink) Subprocess(v obs.Subprocess) {
	s.mu.Lock()
	s.subs = append(s.subs, v)
	full := len(s.subs) >= s.batch
	s.mu.Unlock()
	if full {
		s.Flush()
	}
}

// Flush writes what is buffered. Safe to call at any time and from any
// goroutine; a failure is REMEMBERED rather than returned, because the
// callers are observation funnels that must not care whether analytics
// is healthy.
func (s *Sink) Flush() {
	s.mu.Lock()
	renders, subs := s.renders, s.subs
	s.renders, s.subs = nil, nil
	s.mu.Unlock()
	if len(renders) == 0 && len(subs) == 0 {
		return
	}
	if err := s.write(renders, subs); err != nil {
		s.mu.Lock()
		s.flushErr = err
		s.mu.Unlock()
	}
}

func (s *Sink) write(renders []obs.Render, subs []obs.Subprocess) error {
	tx, err := s.db.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	if len(renders) > 0 {
		st, err := tx.Prepare(`INSERT INTO render
			(ts, path, repo, elapsed_ms, ops_ms, zit_ms, status, url)
			VALUES (?, ?, ?, ?, ?, ?, ?, ?)`)
		if err != nil {
			return err
		}
		for _, v := range renders {
			if _, err := st.Exec(v.At, v.Path, v.Repo,
				v.Elapsed.Milliseconds(), v.Ops.Milliseconds(), v.Zit.Milliseconds(),
				v.Status, v.URL); err != nil {
				st.Close()
				return err
			}
		}
		st.Close()
	}
	if len(subs) > 0 {
		st, err := tx.Prepare(`INSERT INTO subprocess VALUES (?, ?, ?, ?, ?)`)
		if err != nil {
			return err
		}
		for _, v := range subs {
			if _, err := st.Exec(v.At, v.Verb, v.Repo,
				v.Duration.Milliseconds(), v.Err); err != nil {
				st.Close()
				return err
			}
		}
		st.Close()
	}
	return tx.Commit()
}

// Trend reports a route's daily percentiles over the last days, oldest
// first; an empty path trends the whole forge.
//
// quantile_disc, not quantile_cont: the live panel computes percentiles
// by nearest rank, and a history that interpolated would disagree with
// the snapshot above it about the same window. Two numbers on one page
// that are both "p95" and do not match is the kind of small
// contradiction that costs a reader their trust in the rest.
func (s *Sink) Trend(path string, days int) ([]obs.TrendPoint, error) {
	if days <= 0 {
		days = 14
	}
	// Today's rows are the ones an operator is looking for after a
	// deploy, and they are the ones most likely to still be buffered.
	s.Flush()
	where, args := "ts >= now() - INTERVAL (?) DAY", []any{days}
	if path != "" {
		where, args = "path = ? AND "+where, []any{path, days}
	}
	rows, err := s.db.Query(`SELECT strftime(ts, '%Y-%m-%d') AS day, count(*),
			CAST(quantile_disc(elapsed_ms, 0.5) AS BIGINT),
			CAST(quantile_disc(elapsed_ms, 0.95) AS BIGINT),
			CAST(max(elapsed_ms) AS BIGINT)
		FROM render WHERE `+where+`
		GROUP BY day ORDER BY day`, args...)
	if err != nil {
		return nil, fmt.Errorf("obsduck: trend: %w", err)
	}
	defer rows.Close()
	var out []obs.TrendPoint
	for rows.Next() {
		var p obs.TrendPoint
		if err := rows.Scan(&p.Day, &p.N, &p.P50Ms, &p.P95Ms, &p.MaxMs); err != nil {
			return nil, err
		}
		out = append(out, p)
	}
	return out, rows.Err()
}

// History reads renders back, newest or slowest first.
//
// Bounded three ways — a window, a limit, and an optional filter —
// because an unbounded read of an append-only store is the query that
// works fine until the day it matters.
func (s *Sink) History(q obs.HistoryQuery) ([]obs.HistoryRow, error) {
	if q.Days <= 0 {
		q.Days = 14
	}
	if q.Limit <= 0 || q.Limit > 500 {
		q.Limit = 100
	}
	s.Flush()
	where := []string{"ts >= now() - INTERVAL (?) DAY"}
	args := []any{q.Days}
	if q.Find != "" {
		// The route AND the requested path: a reader who types a
		// repository name means the URL, and one who types a route
		// means the pattern. Answering only one of those makes the box
		// look broken half the time.
		where = append(where, "(lower(path) LIKE ? OR lower(coalesce(url, '')) LIKE ?)")
		like := "%" + strings.ToLower(q.Find) + "%"
		args = append(args, like, like)
	}
	if q.Repo != "" {
		where = append(where, "repo = ?")
		args = append(args, q.Repo)
	}
	order := "ts DESC"
	if q.Sort == "slowest" {
		order = "elapsed_ms DESC, ts DESC"
	}
	args = append(args, q.Limit)
	rows, err := s.db.Query(`SELECT strftime(ts, '%Y-%m-%d %H:%M:%S'), path,
			coalesce(url, ''), coalesce(repo, ''), elapsed_ms, ops_ms, zit_ms, status
		FROM render WHERE `+strings.Join(where, " AND ")+`
		ORDER BY `+order+` LIMIT ?`, args...)
	if err != nil {
		return nil, fmt.Errorf("obsduck: history: %w", err)
	}
	defer rows.Close()
	var out []obs.HistoryRow
	for rows.Next() {
		var h obs.HistoryRow
		if err := rows.Scan(&h.At, &h.Path, &h.URL, &h.Repo,
			&h.ElapsedMs, &h.OpsMs, &h.ZitMs, &h.Status); err != nil {
			return nil, err
		}
		out = append(out, h)
	}
	return out, rows.Err()
}

// Prune drops observations older than keep and reports how many rows
// went. A keep of zero or less keeps everything.
//
// Retention is the operator's call, not a default nobody was asked
// about — but "append for ever" is also a default nobody was asked
// about, and it is the one that ends with a forge whose disk fills
// because of the thing that was supposed to be watching it. So there
// is a window, it is stated, and it can be switched off.
func (s *Sink) Prune(keep time.Duration) (int64, error) {
	if keep <= 0 {
		return 0, nil
	}
	// Flush first: rows still in the buffer are the newest ones and
	// would survive anyway, but pruning around an unflushed buffer
	// makes the count reported to an operator a guess.
	s.Flush()
	cutoff := time.Now().Add(-keep)
	var total int64
	for _, table := range []string{"render", "subprocess"} {
		res, err := s.db.Exec(`DELETE FROM `+table+` WHERE ts < ?`, cutoff)
		if err != nil {
			return total, fmt.Errorf("obsduck: prune %s: %w", table, err)
		}
		if n, err := res.RowsAffected(); err == nil {
			total += n
		}
	}
	return total, nil
}

// PruneEvery runs Prune on a timer. Daily is the right cadence for a
// window measured in days: pruning more often costs writes to delete
// rows that a later pass would have deleted anyway.
func (s *Sink) PruneEvery(every, keep time.Duration, stop <-chan struct{}) {
	if every <= 0 {
		every = 24 * time.Hour
	}
	t := time.NewTicker(every)
	defer t.Stop()
	for {
		select {
		case <-t.C:
			if _, err := s.Prune(keep); err != nil {
				s.mu.Lock()
				s.flushErr = err
				s.mu.Unlock()
			}
		case <-stop:
			return
		}
	}
}

// Err reports the last write failure, so an admin surface can say the
// analytics store is unhealthy rather than silently recording nothing.
func (s *Sink) Err() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.flushErr
}

// Close flushes and shuts down.
func (s *Sink) Close() error {
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		return nil
	}
	s.closed = true
	s.mu.Unlock()
	s.Flush()
	return s.db.Close()
}

// FlushEvery drains the buffer on a timer, so a quiet forge still lands
// its rows. Without it a low-traffic instance would hold observations
// in memory until a batch filled, which on a personal forge could be
// days — and a restart would lose exactly the window most worth having.
func (s *Sink) FlushEvery(d time.Duration, stop <-chan struct{}) {
	if d <= 0 {
		d = 30 * time.Second
	}
	t := time.NewTicker(d)
	defer t.Stop()
	for {
		select {
		case <-t.C:
			s.Flush()
		case <-stop:
			s.Flush()
			return
		}
	}
}