initial commit
This commit is contained in:
@@ -0,0 +1,409 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/owen/vibedns/internal/models"
|
||||
)
|
||||
|
||||
// --- Query log ----------------------------------------------------------
|
||||
|
||||
// InsertQueryLogs writes a batch of query log rows in one transaction. The
|
||||
// query logger buffers rows in memory and calls this periodically so DNS
|
||||
// resolution never waits on disk.
|
||||
func (db *DB) InsertQueryLogs(ctx context.Context, entries []models.QueryLogEntry) error {
|
||||
if len(entries) == 0 {
|
||||
return nil
|
||||
}
|
||||
return db.InTx(ctx, func(tx *sql.Tx) error {
|
||||
stmt, err := tx.PrepareContext(ctx, `
|
||||
INSERT INTO query_logs (ts, client_ip, network_id, network_name, qname, qtype, rcode,
|
||||
source, cache_hit, blocked, policy_id, policy_name, blacklist_id, blacklist_name,
|
||||
matched_rule, protocol, duration_us, answer_count)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer stmt.Close()
|
||||
for _, e := range entries {
|
||||
_, err := stmt.ExecContext(ctx,
|
||||
e.Timestamp.UnixMilli(), e.ClientIP, nullInt64(e.NetworkID), e.NetworkName,
|
||||
e.QName, e.QType, e.Rcode, e.Source, boolInt(e.CacheHit), boolInt(e.Blocked),
|
||||
nullInt64(e.PolicyID), e.PolicyName, nullInt64(e.BlacklistID), e.BlacklistName,
|
||||
e.MatchedRule, e.Protocol, e.DurationUS, e.AnswerCount)
|
||||
if err != nil {
|
||||
return fmt.Errorf("write query log: %w", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// QueryLogFilter narrows a query log search.
|
||||
type QueryLogFilter struct {
|
||||
Domain string
|
||||
ClientIP string
|
||||
QType string
|
||||
Rcode string
|
||||
Source string
|
||||
Blocked string // "", "blocked", "allowed"
|
||||
NetworkID int64
|
||||
From time.Time
|
||||
To time.Time
|
||||
Limit int
|
||||
Offset int
|
||||
}
|
||||
|
||||
func (f QueryLogFilter) where() (string, []any) {
|
||||
var conds []string
|
||||
var args []any
|
||||
if s := strings.TrimSpace(f.Domain); s != "" {
|
||||
conds = append(conds, "qname LIKE ?")
|
||||
args = append(args, "%"+strings.ToLower(s)+"%")
|
||||
}
|
||||
if s := strings.TrimSpace(f.ClientIP); s != "" {
|
||||
conds = append(conds, "client_ip LIKE ?")
|
||||
args = append(args, "%"+s+"%")
|
||||
}
|
||||
if s := strings.ToUpper(strings.TrimSpace(f.QType)); s != "" {
|
||||
conds = append(conds, "qtype = ?")
|
||||
args = append(args, s)
|
||||
}
|
||||
if s := strings.ToUpper(strings.TrimSpace(f.Rcode)); s != "" {
|
||||
conds = append(conds, "rcode = ?")
|
||||
args = append(args, s)
|
||||
}
|
||||
if s := strings.TrimSpace(f.Source); s != "" {
|
||||
conds = append(conds, "source = ?")
|
||||
args = append(args, s)
|
||||
}
|
||||
switch f.Blocked {
|
||||
case "blocked":
|
||||
conds = append(conds, "blocked = 1")
|
||||
case "allowed":
|
||||
conds = append(conds, "blocked = 0")
|
||||
}
|
||||
if f.NetworkID > 0 {
|
||||
conds = append(conds, "network_id = ?")
|
||||
args = append(args, f.NetworkID)
|
||||
}
|
||||
if !f.From.IsZero() {
|
||||
conds = append(conds, "ts >= ?")
|
||||
args = append(args, f.From.UnixMilli())
|
||||
}
|
||||
if !f.To.IsZero() {
|
||||
conds = append(conds, "ts <= ?")
|
||||
args = append(args, f.To.UnixMilli())
|
||||
}
|
||||
if len(conds) == 0 {
|
||||
return "", nil
|
||||
}
|
||||
return " WHERE " + strings.Join(conds, " AND "), args
|
||||
}
|
||||
|
||||
// QueryLogs returns matching rows newest first, plus the total match count.
|
||||
func (db *DB) QueryLogs(ctx context.Context, f QueryLogFilter) ([]models.QueryLogEntry, int, error) {
|
||||
whereSQL, args := f.where()
|
||||
|
||||
var total int
|
||||
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM query_logs`+whereSQL, args...).Scan(&total); err != nil {
|
||||
return nil, 0, fmt.Errorf("count query logs: %w", err)
|
||||
}
|
||||
|
||||
q := `SELECT id, ts, client_ip, network_id, network_name, qname, qtype, rcode, source,
|
||||
cache_hit, blocked, policy_id, policy_name, blacklist_id, blacklist_name, matched_rule,
|
||||
protocol, duration_us, answer_count
|
||||
FROM query_logs` + whereSQL + ` ORDER BY ts DESC, id DESC`
|
||||
qargs := args
|
||||
if f.Limit > 0 {
|
||||
q += ` LIMIT ? OFFSET ?`
|
||||
qargs = append(append([]any{}, args...), f.Limit, f.Offset)
|
||||
}
|
||||
|
||||
rows, err := db.QueryContext(ctx, q, qargs...)
|
||||
if err != nil {
|
||||
return nil, 0, fmt.Errorf("read query logs: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []models.QueryLogEntry
|
||||
for rows.Next() {
|
||||
var e models.QueryLogEntry
|
||||
var ts int64
|
||||
var netID, polID, blID sql.NullInt64
|
||||
var cacheHit, blocked int
|
||||
err := rows.Scan(&e.ID, &ts, &e.ClientIP, &netID, &e.NetworkName, &e.QName, &e.QType,
|
||||
&e.Rcode, &e.Source, &cacheHit, &blocked, &polID, &e.PolicyName, &blID,
|
||||
&e.BlacklistName, &e.MatchedRule, &e.Protocol, &e.DurationUS, &e.AnswerCount)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
e.Timestamp = time.UnixMilli(ts)
|
||||
e.CacheHit = cacheHit != 0
|
||||
e.Blocked = blocked != 0
|
||||
if netID.Valid {
|
||||
v := netID.Int64
|
||||
e.NetworkID = &v
|
||||
}
|
||||
if polID.Valid {
|
||||
v := polID.Int64
|
||||
e.PolicyID = &v
|
||||
}
|
||||
if blID.Valid {
|
||||
v := blID.Int64
|
||||
e.BlacklistID = &v
|
||||
}
|
||||
out = append(out, e)
|
||||
}
|
||||
return out, total, rows.Err()
|
||||
}
|
||||
|
||||
// PruneQueryLogs enforces the retention policy. Rows older than retentionDays
|
||||
// are removed first, then the table is trimmed to maxRows newest entries.
|
||||
// Either limit may be zero to disable it.
|
||||
func (db *DB) PruneQueryLogs(ctx context.Context, retentionDays, maxRows int) (int64, error) {
|
||||
var deleted int64
|
||||
if retentionDays > 0 {
|
||||
cutoff := time.Now().AddDate(0, 0, -retentionDays).UnixMilli()
|
||||
res, err := db.ExecContext(ctx, `DELETE FROM query_logs WHERE ts < ?`, cutoff)
|
||||
if err != nil {
|
||||
return deleted, fmt.Errorf("prune query logs by age: %w", err)
|
||||
}
|
||||
n, _ := res.RowsAffected()
|
||||
deleted += n
|
||||
}
|
||||
if maxRows > 0 {
|
||||
res, err := db.ExecContext(ctx, `
|
||||
DELETE FROM query_logs WHERE id NOT IN (
|
||||
SELECT id FROM query_logs ORDER BY ts DESC, id DESC LIMIT ?
|
||||
)`, maxRows)
|
||||
if err != nil {
|
||||
return deleted, fmt.Errorf("prune query logs by count: %w", err)
|
||||
}
|
||||
n, _ := res.RowsAffected()
|
||||
deleted += n
|
||||
}
|
||||
return deleted, nil
|
||||
}
|
||||
|
||||
// TruncateQueryLogs empties the query log table.
|
||||
func (db *DB) TruncateQueryLogs(ctx context.Context) (int64, error) {
|
||||
res, err := db.ExecContext(ctx, `DELETE FROM query_logs`)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("clear query logs: %w", err)
|
||||
}
|
||||
n, _ := res.RowsAffected()
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// NameCount is a domain/client aggregate used by the dashboard top-N lists.
|
||||
type NameCount struct {
|
||||
Name string `json:"name"`
|
||||
Count int64 `json:"count"`
|
||||
Extra string `json:"extra,omitempty"`
|
||||
}
|
||||
|
||||
// TopQueried returns the most frequently queried names since `since`.
|
||||
func (db *DB) TopQueried(ctx context.Context, since time.Time, limit int) ([]NameCount, error) {
|
||||
return db.topBy(ctx, `SELECT qname, COUNT(*) c FROM query_logs WHERE ts >= ? GROUP BY qname ORDER BY c DESC LIMIT ?`,
|
||||
since.UnixMilli(), limit)
|
||||
}
|
||||
|
||||
// TopBlocked returns the most frequently blocked names since `since`.
|
||||
func (db *DB) TopBlocked(ctx context.Context, since time.Time, limit int) ([]NameCount, error) {
|
||||
return db.topBy(ctx, `SELECT qname, COUNT(*) c FROM query_logs WHERE blocked = 1 AND ts >= ? GROUP BY qname ORDER BY c DESC LIMIT ?`,
|
||||
since.UnixMilli(), limit)
|
||||
}
|
||||
|
||||
// TopClients returns the busiest clients since `since`.
|
||||
func (db *DB) TopClients(ctx context.Context, since time.Time, limit int) ([]NameCount, error) {
|
||||
rows, err := db.QueryContext(ctx, `
|
||||
SELECT client_ip, COUNT(*) c, COALESCE(MAX(network_name), '')
|
||||
FROM query_logs WHERE ts >= ? GROUP BY client_ip ORDER BY c DESC LIMIT ?`,
|
||||
since.UnixMilli(), limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("top clients: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []NameCount
|
||||
for rows.Next() {
|
||||
var n NameCount
|
||||
if err := rows.Scan(&n.Name, &n.Count, &n.Extra); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, n)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (db *DB) topBy(ctx context.Context, q string, args ...any) ([]NameCount, error) {
|
||||
rows, err := db.QueryContext(ctx, q, args...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("aggregate query logs: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []NameCount
|
||||
for rows.Next() {
|
||||
var n NameCount
|
||||
if err := rows.Scan(&n.Name, &n.Count); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, n)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// TimeBucket is one point on the dashboard activity chart.
|
||||
type TimeBucket struct {
|
||||
Start time.Time `json:"start"`
|
||||
Total int64 `json:"total"`
|
||||
Blocked int64 `json:"blocked"`
|
||||
Cached int64 `json:"cached"`
|
||||
}
|
||||
|
||||
// ActivityBuckets groups query log rows into fixed-width time buckets covering
|
||||
// the window [since, now].
|
||||
func (db *DB) ActivityBuckets(ctx context.Context, since time.Time, bucket time.Duration, count int) ([]TimeBucket, error) {
|
||||
if bucket <= 0 || count <= 0 {
|
||||
return nil, nil
|
||||
}
|
||||
width := bucket.Milliseconds()
|
||||
start := since.UnixMilli()
|
||||
|
||||
rows, err := db.QueryContext(ctx, `
|
||||
SELECT (ts - ?) / ? AS b, COUNT(*), SUM(blocked), SUM(cache_hit)
|
||||
FROM query_logs WHERE ts >= ?
|
||||
GROUP BY b ORDER BY b`, start, width, start)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("bucket query logs: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
buckets := make([]TimeBucket, count)
|
||||
for i := range buckets {
|
||||
buckets[i].Start = time.UnixMilli(start + int64(i)*width)
|
||||
}
|
||||
for rows.Next() {
|
||||
var idx, total int64
|
||||
var blocked, cached sql.NullInt64
|
||||
if err := rows.Scan(&idx, &total, &blocked, &cached); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if idx < 0 || idx >= int64(count) {
|
||||
continue
|
||||
}
|
||||
buckets[idx].Total = total
|
||||
buckets[idx].Blocked = blocked.Int64
|
||||
buckets[idx].Cached = cached.Int64
|
||||
}
|
||||
return buckets, rows.Err()
|
||||
}
|
||||
|
||||
// QueryLogCount returns the number of stored rows.
|
||||
func (db *DB) QueryLogCount(ctx context.Context) (int64, error) {
|
||||
var n int64
|
||||
err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM query_logs`).Scan(&n)
|
||||
return n, err
|
||||
}
|
||||
|
||||
// --- Audit log ----------------------------------------------------------
|
||||
|
||||
// InsertAudit appends an administrative audit entry.
|
||||
func (db *DB) InsertAudit(ctx context.Context, e models.AuditEntry) error {
|
||||
if e.Timestamp.IsZero() {
|
||||
e.Timestamp = time.Now()
|
||||
}
|
||||
_, err := db.ExecContext(ctx, `
|
||||
INSERT INTO audit_logs (ts, actor, source, client_ip, action, object_type, object_id, object_name, details)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
e.Timestamp.UnixMilli(), e.Actor, e.Source, e.ClientIP, e.Action,
|
||||
e.ObjectType, e.ObjectID, e.ObjectName, e.Details)
|
||||
if err != nil {
|
||||
return fmt.Errorf("write audit log: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// AuditFilter narrows an audit log search.
|
||||
type AuditFilter struct {
|
||||
Search string
|
||||
ObjectType string
|
||||
Source string
|
||||
Limit int
|
||||
Offset int
|
||||
}
|
||||
|
||||
// AuditLogs returns audit entries newest first, plus the total match count.
|
||||
func (db *DB) AuditLogs(ctx context.Context, f AuditFilter) ([]models.AuditEntry, int, error) {
|
||||
var conds []string
|
||||
var args []any
|
||||
if s := strings.TrimSpace(f.Search); s != "" {
|
||||
conds = append(conds, "(action LIKE ? OR object_name LIKE ? OR details LIKE ? OR actor LIKE ?)")
|
||||
pat := "%" + s + "%"
|
||||
args = append(args, pat, pat, pat, pat)
|
||||
}
|
||||
if f.ObjectType != "" {
|
||||
conds = append(conds, "object_type = ?")
|
||||
args = append(args, f.ObjectType)
|
||||
}
|
||||
if f.Source != "" {
|
||||
conds = append(conds, "source = ?")
|
||||
args = append(args, f.Source)
|
||||
}
|
||||
whereSQL := ""
|
||||
if len(conds) > 0 {
|
||||
whereSQL = " WHERE " + strings.Join(conds, " AND ")
|
||||
}
|
||||
|
||||
var total int
|
||||
if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM audit_logs`+whereSQL, args...).Scan(&total); err != nil {
|
||||
return nil, 0, fmt.Errorf("count audit logs: %w", err)
|
||||
}
|
||||
|
||||
q := `SELECT id, ts, actor, source, client_ip, action, object_type, object_id, object_name, details
|
||||
FROM audit_logs` + whereSQL + ` ORDER BY ts DESC, id DESC`
|
||||
qargs := args
|
||||
if f.Limit > 0 {
|
||||
q += ` LIMIT ? OFFSET ?`
|
||||
qargs = append(append([]any{}, args...), f.Limit, f.Offset)
|
||||
}
|
||||
|
||||
rows, err := db.QueryContext(ctx, q, qargs...)
|
||||
if err != nil {
|
||||
return nil, 0, fmt.Errorf("read audit logs: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []models.AuditEntry
|
||||
for rows.Next() {
|
||||
var e models.AuditEntry
|
||||
var ts int64
|
||||
if err := rows.Scan(&e.ID, &ts, &e.Actor, &e.Source, &e.ClientIP, &e.Action,
|
||||
&e.ObjectType, &e.ObjectID, &e.ObjectName, &e.Details); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
e.Timestamp = time.UnixMilli(ts)
|
||||
out = append(out, e)
|
||||
}
|
||||
return out, total, rows.Err()
|
||||
}
|
||||
|
||||
// PruneAuditLogs trims the audit log to the newest maxRows entries.
|
||||
func (db *DB) PruneAuditLogs(ctx context.Context, maxRows int) (int64, error) {
|
||||
if maxRows <= 0 {
|
||||
return 0, nil
|
||||
}
|
||||
res, err := db.ExecContext(ctx, `
|
||||
DELETE FROM audit_logs WHERE id NOT IN (
|
||||
SELECT id FROM audit_logs ORDER BY ts DESC, id DESC LIMIT ?
|
||||
)`, maxRows)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("prune audit logs: %w", err)
|
||||
}
|
||||
n, _ := res.RowsAffected()
|
||||
return n, nil
|
||||
}
|
||||
Reference in New Issue
Block a user