// Reports that arrive from devices over MQTT: presence and telemetry // (AGENTS.md §7, §11 Phase 3). // // The broker is transport only. Every report is written to SQLite here, because // SQLite is the source of truth (SYSTEM_MAP.md). A malformed or hostile report // must never take the service down: callers validate first, and a failed insert // is logged and dropped. package store import ( "context" "database/sql" "fmt" "time" ) // SetPresence records that a device was seen and whether it is online now. It // returns whether the online flag changed. // // Presence comes from the broker's Last Will and Testament: a device publishes // a retained "online" status, and the broker publishes its retained will // ("online": false) when the connection drops without a clean disconnect. // // last_seen is refreshed on every report. The audit log only gets an entry when // the online flag changes, so a chatty device does not flood the log (FR-15). func (s *Store) SetPresence(ctx context.Context, id int64, online bool, at time.Time, actor string) (bool, error) { if at.IsZero() { at = time.Now() } at = at.UTC() tx, err := s.db.BeginTx(ctx, nil) if err != nil { return false, fmt.Errorf("set presence: %w", err) } defer func() { _ = tx.Rollback() }() var current int if err := tx.QueryRowContext(ctx, `SELECT online FROM devices WHERE id = ?`, id).Scan(¤t); err != nil { return false, fmt.Errorf("set presence: %w", err) } if _, err := tx.ExecContext(ctx, `UPDATE devices SET online = ?, last_seen = ?, updated_at = datetime('now') WHERE id = ?`, boolToInt(online), at.Format(time.RFC3339), id); err != nil { return false, fmt.Errorf("set presence: %w", err) } changed := (current != 0) != online if changed { action, detail := "device.offline", "last will received" if online { action, detail = "device.online", "status report" } if err := insertAudit(ctx, tx, actor, action, &id, detail); err != nil { return false, err } } if err := tx.Commit(); err != nil { return false, fmt.Errorf("commit presence: %w", err) } return changed, nil } // TelemetryEntry is one app or domain row of a usage report. type TelemetryEntry struct { App string Domain string Seconds int } // InsertTelemetry upserts a rolling daily usage total per (device, day, app, // domain). Reports are repeats, not deltas: a later report for the same key // replaces the earlier total (migration 002). The affected days are rolled up // in the same transaction (migration 003). It returns the number of rows // written. func (s *Store) InsertTelemetry(ctx context.Context, id int64, day string, entries []TelemetryEntry) (int, error) { if len(entries) == 0 { return 0, nil } tx, err := s.db.BeginTx(ctx, nil) if err != nil { return 0, fmt.Errorf("insert telemetry: %w", err) } defer func() { _ = tx.Rollback() }() n := 0 days := make(map[string]struct{}, 1) for _, e := range entries { if e.Seconds < 0 { continue } if _, err := tx.ExecContext(ctx, ` INSERT INTO telemetry (device_id, app, domain, seconds, day) VALUES (?, ?, ?, ?, ?) ON CONFLICT(device_id, day, app, domain) DO UPDATE SET seconds = excluded.seconds, created_at = datetime('now')`, id, e.App, e.Domain, e.Seconds, day); err != nil { return n, fmt.Errorf("insert telemetry: %w", err) } days[day] = struct{}{} n++ } for d := range days { if err := rollupDay(ctx, tx, id, d); err != nil { return n, err } } if err := tx.Commit(); err != nil { return n, fmt.Errorf("commit telemetry: %w", err) } return n, nil } // rollupDay recomputes one device/day row of telemetry_daily from the detail // table. Apps and domains are summed separately so a report that carries both // is not double-counted. func rollupDay(ctx context.Context, ex interface { ExecContext(context.Context, string, ...any) (sql.Result, error) }, id int64, day string) error { if _, err := ex.ExecContext(ctx, ` INSERT INTO telemetry_daily (device_id, day, app_seconds, domain_seconds, app_count, domain_count, updated_at) SELECT ?, ?, COALESCE(SUM(CASE WHEN app <> '' THEN seconds END), 0), COALESCE(SUM(CASE WHEN domain <> '' THEN seconds END), 0), COALESCE(SUM(CASE WHEN app <> '' THEN 1 END), 0), COALESCE(SUM(CASE WHEN domain <> '' THEN 1 END), 0), datetime('now') FROM telemetry WHERE device_id = ? AND day = ? ON CONFLICT(device_id, day) DO UPDATE SET app_seconds = excluded.app_seconds, domain_seconds = excluded.domain_seconds, app_count = excluded.app_count, domain_count = excluded.domain_count, updated_at = datetime('now')`, id, day, id, day); err != nil { return fmt.Errorf("rollup day: %w", err) } return nil } // RebuildRollups recomputes every telemetry_daily row from the detail table. It // is idempotent and is how the rollup is backfilled after an upgrade or // repaired after an import. It returns the number of rows inserted or updated. func (s *Store) RebuildRollups(ctx context.Context) (int, error) { res, err := s.db.ExecContext(ctx, ` INSERT INTO telemetry_daily (device_id, day, app_seconds, domain_seconds, app_count, domain_count, updated_at) SELECT device_id, day, COALESCE(SUM(CASE WHEN app <> '' THEN seconds END), 0), COALESCE(SUM(CASE WHEN domain <> '' THEN seconds END), 0), COALESCE(SUM(CASE WHEN app <> '' THEN 1 END), 0), COALESCE(SUM(CASE WHEN domain <> '' THEN 1 END), 0), datetime('now') FROM telemetry GROUP BY device_id, day ON CONFLICT(device_id, day) DO UPDATE SET app_seconds = excluded.app_seconds, domain_seconds = excluded.domain_seconds, app_count = excluded.app_count, domain_count = excluded.domain_count, updated_at = datetime('now')`) if err != nil { return 0, fmt.Errorf("rebuild rollups: %w", err) } n, err := res.RowsAffected() if err != nil { return 0, fmt.Errorf("rebuild rollups: %w", err) } return int(n), nil } // DayUsage is one day of rolled-up usage for a device. AppSeconds and // DomainSeconds are kept apart because a report may carry both. type DayUsage struct { Day string AppSeconds int DomainSeconds int AppCount int DomainCount int } // UsageRange returns the daily rollup for a device in [from, to] inclusive, // oldest first. Days without a report are absent; the caller fills the gaps. func (s *Store) UsageRange(ctx context.Context, id int64, from, to string) ([]DayUsage, error) { rows, err := s.db.QueryContext(ctx, ` SELECT day, app_seconds, domain_seconds, app_count, domain_count FROM telemetry_daily WHERE device_id = ? AND day BETWEEN ? AND ? ORDER BY day`, id, from, to) if err != nil { return nil, fmt.Errorf("usage range: %w", err) } defer rows.Close() out := make([]DayUsage, 0, 8) for rows.Next() { var d DayUsage if err := rows.Scan(&d.Day, &d.AppSeconds, &d.DomainSeconds, &d.AppCount, &d.DomainCount); err != nil { return nil, fmt.Errorf("scan usage day: %w", err) } out = append(out, d) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("usage range: %w", err) } return out, nil } // UsageTotal is one app or domain total over a range. type UsageTotal struct { Name string Seconds int } // TopUsage returns the biggest apps or domains for a device in [from, to] // inclusive. kind is "app" or "domain"; anything else returns no rows. The // column is chosen by a switch, never interpolated, so the argument can never // reach SQL as text. func (s *Store) TopUsage(ctx context.Context, id int64, from, to, kind string, limit int) ([]UsageTotal, error) { column := "" switch kind { case "app": column = "app" case "domain": column = "domain" default: return nil, nil } if limit <= 0 || limit > 100 { limit = 10 } rows, err := s.db.QueryContext(ctx, ` SELECT `+column+`, SUM(seconds) FROM telemetry WHERE device_id = ? AND day BETWEEN ? AND ? AND `+column+` <> '' GROUP BY `+column+` ORDER BY SUM(seconds) DESC, `+column+` ASC LIMIT ?`, id, from, to, limit) if err != nil { return nil, fmt.Errorf("top %s: %w", column, err) } defer rows.Close() out := make([]UsageTotal, 0, limit) for rows.Next() { var t UsageTotal if err := rows.Scan(&t.Name, &t.Seconds); err != nil { return nil, fmt.Errorf("scan top %s: %w", column, err) } out = append(out, t) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("top %s: %w", column, err) } return out, nil } // DeviceDayUsage is one device's rolled-up usage for one day. AppSeconds and // DomainSeconds are kept apart for the same reason as the rollup: a report may // carry both for one session, so adding them together would double-count time. // The screen-time answer decides how to combine them (internal/hub). type DeviceDayUsage struct { DeviceID int64 Name string Child string AppSeconds int DomainSeconds int } // UsageByDeviceForDay returns the daily rollup for every device that reported // anything on day (YYYY-MM-DD). It is the read the screen-time answer is built // from. Devices with no row are absent; the caller treats them as zero. func (s *Store) UsageByDeviceForDay(ctx context.Context, day string) ([]DeviceDayUsage, error) { rows, err := s.db.QueryContext(ctx, ` SELECT d.id, d.name, d.child, td.app_seconds, td.domain_seconds FROM telemetry_daily td JOIN devices d ON d.id = td.device_id WHERE td.day = ? ORDER BY d.id`, day) if err != nil { return nil, fmt.Errorf("usage by device for day: %w", err) } defer rows.Close() out := make([]DeviceDayUsage, 0, 8) for rows.Next() { var u DeviceDayUsage if err := rows.Scan(&u.DeviceID, &u.Name, &u.Child, &u.AppSeconds, &u.DomainSeconds); err != nil { return nil, fmt.Errorf("scan device day usage: %w", err) } out = append(out, u) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("usage by device for day: %w", err) } return out, nil } // DetailUsageByDeviceForDay is the same measure computed straight from the // telemetry detail table instead of the derived rollup. It exists so an // operator can independently check the rollup the daily answer reads. func (s *Store) DetailUsageByDeviceForDay(ctx context.Context, day string) ([]DeviceDayUsage, error) { rows, err := s.db.QueryContext(ctx, ` SELECT d.id, d.name, d.child, COALESCE(SUM(CASE WHEN t.app <> '' THEN t.seconds END), 0), COALESCE(SUM(CASE WHEN t.domain <> '' THEN t.seconds END), 0) FROM telemetry t JOIN devices d ON d.id = t.device_id WHERE t.day = ? GROUP BY d.id ORDER BY d.id`, day) if err != nil { return nil, fmt.Errorf("detail usage by device for day: %w", err) } defer rows.Close() out := make([]DeviceDayUsage, 0, 8) for rows.Next() { var u DeviceDayUsage if err := rows.Scan(&u.DeviceID, &u.Name, &u.Child, &u.AppSeconds, &u.DomainSeconds); err != nil { return nil, fmt.Errorf("scan detail device day usage: %w", err) } out = append(out, u) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("detail usage by device for day: %w", err) } return out, nil } // CapAudit keeps the newest max audit rows and deletes the oldest beyond that, // so the append-only log cannot grow without bound (AGENTS.md §7 retention). It // returns how many rows it removed. Nothing in the audit log is ever edited. func (s *Store) CapAudit(ctx context.Context, max int) (int, error) { if max <= 0 { return 0, nil } res, err := s.db.ExecContext(ctx, ` DELETE FROM audit WHERE id NOT IN ( SELECT id FROM audit ORDER BY id DESC LIMIT ? )`, max) if err != nil { return 0, fmt.Errorf("cap audit: %w", err) } n, err := res.RowsAffected() if err != nil { return 0, fmt.Errorf("cap audit: %w", err) } return int(n), nil } // TelemetryForDay returns one device's usage rows for a day, biggest first. func (s *Store) TelemetryForDay(ctx context.Context, id int64, day string) ([]TelemetryEntry, error) { rows, err := s.db.QueryContext(ctx, ` SELECT app, domain, seconds FROM telemetry WHERE device_id = ? AND day = ? ORDER BY seconds DESC`, id, day) if err != nil { return nil, fmt.Errorf("telemetry for day: %w", err) } defer rows.Close() out := make([]TelemetryEntry, 0, 8) for rows.Next() { var e TelemetryEntry if err := rows.Scan(&e.App, &e.Domain, &e.Seconds); err != nil { return nil, fmt.Errorf("scan telemetry: %w", err) } out = append(out, e) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("telemetry for day: %w", err) } return out, nil }