//go:build !js && !wasip1
package obsduck
import (
"context"
"fmt"
"time"
"craftdesign.group/quarry/internal/obs"
)
// swarmSchema is the swarm's own table: no address, no visitor, only
// what requests of a distributed crawler share (obs.Swarm).
const swarmSchema = `CREATE TABLE IF NOT EXISTS swarm (
ts TIMESTAMP, prefix VARCHAR, asn UINTEGER, network VARCHAR, country VARCHAR,
ua VARCHAR, hints VARCHAR, lang VARCHAR, headset VARCHAR,
shape VARCHAR, repo VARCHAR, status INTEGER)`
// sweepSwarm keeps, before their addresses are blanked, the hits a
// distributed crawler sent: a visitor's only request, through the web,
// signed in as nobody, for a page named by a hash, and one Classify
// would call a person. Each such row keeps its range (from the address,
// so rows filed before prefixes were recorded count too) and how it
// presented itself. The address and the visitor are not kept.
func (s *Sink) sweepSwarm(ctx context.Context, cutoff time.Time) (int64, error) {
rows, err := s.db.QueryContext(ctx, `
WITH c AS (
SELECT ts, addr, visitor, coalesce(asn, CAST(0 AS UINTEGER)) AS asn, coalesce(network, '') AS network,
coalesce(country, '') AS country, coalesce(ua, '') AS ua, coalesce(client, '') AS client,
coalesce(hints, '') AS hints, coalesce(lang, '') AS lang, coalesce(headset, '') AS headset,
url, coalesce(repo, '') AS repo, coalesce(status, 0) AS status
FROM hit
WHERE addr IS NOT NULL AND ts < ? AND door = 'web' AND coalesce(account, '') = ''
AND regexp_matches(url, ?)),
one AS (
SELECT visitor FROM hit WHERE visitor IN (SELECT visitor FROM c)
GROUP BY visitor HAVING count(*) = 1)
SELECT ts, addr, asn, network, country, ua, client, hints, lang, headset, url, repo, status
FROM c JOIN one USING (visitor)`, cutoff, obs.HashPage.String())
if err != nil {
return 0, err
}
type row struct {
ts time.Time
addr, network, country, ua, client, hints, lang, set string
url, repo string
asn uint32
status int
}
var keep []row
for rows.Next() {
var r row
if err := rows.Scan(&r.ts, &r.addr, &r.asn, &r.network, &r.country, &r.ua, &r.client,
&r.hints, &r.lang, &r.set, &r.url, &r.repo, &r.status); err != nil {
rows.Close()
return 0, err
}
// What the traffic page would have called a person: the requests
// this table exists to catch.
if obs.Classify(obs.Facts{UA: r.ua, Door: obs.DoorWeb, Client: r.client, ASN: r.asn}).Kind == obs.KindPerson {
keep = append(keep, r)
}
}
rows.Close()
if err := rows.Err(); err != nil {
return 0, err
}
if len(keep) == 0 {
return 0, nil
}
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return 0, err
}
defer tx.Rollback()
st, err := tx.PrepareContext(ctx, `INSERT INTO swarm
(ts, prefix, asn, network, country, ua, hints, lang, headset, shape, repo, status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
if err != nil {
return 0, err
}
defer st.Close()
for _, r := range keep {
if _, err := st.ExecContext(ctx, r.ts, obs.PrefixOf(r.addr), r.asn, r.network, r.country,
r.ua, r.hints, r.lang, r.set, obs.PathShape(r.url), r.repo, r.status); err != nil {
return 0, err
}
}
return int64(len(keep)), tx.Commit()
}
// Swarm reads the swarm's rows for the last days.
func (s *Sink) Swarm(now time.Time, days int) (*obs.Swarm, error) {
if days <= 0 || days > 90 {
days = 7
}
ctx, cancel := context.WithTimeout(s.ctx, 20*time.Second)
defer cancel()
from := now.Add(-time.Duration(days) * 24 * time.Hour)
out := &obs.Swarm{From: from, To: now}
if err := s.db.QueryRowContext(ctx, `SELECT count(*) FROM swarm WHERE ts >= ?`, from).Scan(&out.Requests); err != nil {
return nil, fmt.Errorf("obsduck: swarm: %w", err)
}
series := func(unit string, into *[]obs.SwarmDay) error {
rows, err := s.db.QueryContext(ctx, `SELECT date_trunc('`+unit+`', ts) AS t, count(*),
count(DISTINCT prefix), count(DISTINCT asn) FROM swarm WHERE ts >= ? GROUP BY 1 ORDER BY 1`, from)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var d obs.SwarmDay
if err := rows.Scan(&d.At, &d.Requests, &d.Prefixes, &d.Networks); err != nil {
return err
}
*into = append(*into, d)
}
return rows.Err()
}
if err := series("day", &out.Days); err != nil {
return nil, fmt.Errorf("obsduck: swarm days: %w", err)
}
if err := series("hour", &out.Hours); err != nil {
return nil, fmt.Errorf("obsduck: swarm hours: %w", err)
}
rows, err := s.db.QueryContext(ctx, `SELECT prefix, any_value(asn), any_value(network), any_value(country),
count(*), count(DISTINCT date_trunc('day', ts)), min(ts), max(ts)
FROM swarm WHERE ts >= ? AND prefix <> '' GROUP BY prefix ORDER BY 5 DESC, 1 LIMIT 200`, from)
if err != nil {
return nil, fmt.Errorf("obsduck: swarm prefixes: %w", err)
}
for rows.Next() {
var p obs.SwarmPrefix
if err := rows.Scan(&p.Prefix, &p.ASN, &p.Network, &p.Country, &p.Requests, &p.Days, &p.First, &p.Last); err != nil {
rows.Close()
return nil, err
}
out.Prefixes = append(out.Prefixes, p)
}
rows.Close()
rows, err = s.db.QueryContext(ctx, `SELECT asn, any_value(network), count(*), count(DISTINCT prefix)
FROM swarm WHERE ts >= ? GROUP BY asn ORDER BY 3 DESC LIMIT 50`, from)
if err != nil {
return nil, fmt.Errorf("obsduck: swarm networks: %w", err)
}
for rows.Next() {
var n obs.SwarmNetwork
if err := rows.Scan(&n.ASN, &n.Network, &n.Requests, &n.Prefixes); err != nil {
rows.Close()
return nil, err
}
out.Networks = append(out.Networks, n)
}
rows.Close()
counts := func(col string, limit int, into *[]obs.SwarmCount) error {
rows, err := s.db.QueryContext(ctx, `SELECT coalesce(`+col+`, ''), count(*) FROM swarm WHERE ts >= ?
GROUP BY 1 ORDER BY 2 DESC, 1 LIMIT `+fmt.Sprint(limit), from)
if err != nil {
return err
}
defer rows.Close()
for rows.Next() {
var c obs.SwarmCount
if err := rows.Scan(&c.Key, &c.Count); err != nil {
return err
}
*into = append(*into, c)
}
return rows.Err()
}
for _, c := range []struct {
col string
limit int
into *[]obs.SwarmCount
}{
{"hints", 20, &out.Hints}, {"lang", 20, &out.Langs}, {"headset", 20, &out.HeadSets},
{"shape", 20, &out.Shapes}, {"repo", 20, &out.Repos}, {"ua", 20, &out.Agents}, {"country", 30, &out.Countries},
} {
if err := counts(c.col, c.limit, c.into); err != nil {
return nil, fmt.Errorf("obsduck: swarm %s: %w", c.col, err)
}
}
return out, nil
}